{
    "file_diffs": [
        {
            "old_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Provide access to stuff (html, data files) via HTTP and HTTPS\n\n\"\"\"\nimport re\nimport requests\nimport requests.auth\n# at some point was trying to be too specific about which exceptions to\n# catch for a retry of a download.\n# from urllib3.exceptions import MaxRetryError, NewConnectionError\n\nimport io\nfrom time import sleep\n\nfrom ..utils import assure_list_from_str, assure_dict_from_str\nfrom ..dochelpers import borrowkwargs\n\nfrom ..ui import ui\nfrom ..utils import auto_repr\nfrom ..dochelpers import exc_str\nfrom ..support.network import get_url_filename\nfrom ..support.network import get_response_disposition_filename\nfrom ..support.network import rfc2822_to_epoch\nfrom ..support.cookies import cookies_db\nfrom ..support.status import FileStatus\nfrom ..support.exceptions import (\n    DownloadError,\n    AccessDeniedError,\n    AccessFailedError,\n    UnhandledRedirectError,\n)\n\nfrom .base import Authenticator\nfrom .base import BaseDownloader, DownloaderSession\n\nfrom logging import getLogger\nfrom ..log import LoggerHelper\nlgr = getLogger('datalad.http')\n\ntry:\n    import requests_ftp\n    requests_ftp.monkeypatch_session()\nexcept ImportError as e:\n    lgr.debug(\"Failed to import requests_ftp, thus no ftp support: %s\" % exc_str(e))\n    pass\n\nif lgr.getEffectiveLevel() <= 1:\n    # Let's also enable requests etc debugging\n\n    # These two lines enable debugging at httplib level (requests->urllib3->http.client)\n    # You will see the REQUEST, including HEADERS and DATA, and RESPONSE with HEADERS but without DATA.\n    # The only thing missing will be the response.body which is not logged.\n    import http.client\n    # TODO: nohow wrapped with logging, plain prints (heh heh), so formatting will not be consistent\n    http.client.HTTPConnection.debuglevel = 1\n\n    # for requests we can define logging properly\n    requests_log = LoggerHelper(logtarget=\"requests.packages.urllib3\").get_initialized_logger()\n    requests_log.setLevel(lgr.getEffectiveLevel())\n    requests_log.propagate = True\n\n__docformat__ = 'restructuredtext'\n\n\ndef process_www_authenticate(v):\n    if not v:\n        return []\n    supported_type = v.split(' ')[0].lower()\n    our_type = {\n        'basic': 'http_basic_auth',\n        'digest': 'http_digest_auth',\n    }.get(supported_type)\n    return [our_type] if our_type else []\n\n\ndef check_response_status(response, err_prefix=\"\", session=None):\n    \"\"\"Check if response's status_code signals problem with authentication etc\n\n    ATM succeeds only if response code was 200\n    \"\"\"\n    if not err_prefix:\n        err_prefix = \"Access to %s has failed: \" % response.url\n    # 401 would be for digest authentication mechanism, or if we first ask which mechanisms are\n    # supported.... must be linked into the logic if we decide to automagically detect which\n    # mechanism or to give more sensible error message\n    err_msg = err_prefix + \"status code %d\" % response.status_code\n    if response.status_code in {404}:\n        # It could have been that form_url is wrong, so let's just say that\n        # TODO: actually may be that is where we could use tagid and actually determine the form submission url\n        raise DownloadError(err_prefix + \"not found\")\n    elif 400 <= response.status_code < 500:\n        raise AccessDeniedError(\n            err_msg,\n            supported_types=process_www_authenticate(\n                response.headers.get('WWW-Authenticate')))\n    elif response.status_code in {200}:\n        pass\n    elif response.status_code in {301, 302, 307}:\n        # TODO: apparently tests do not excercise this one yet\n        if session is None:\n            raise AccessFailedError(err_msg + \" no session was provided\")\n        redirs = list(session.resolve_redirects(response, response.request))\n        if len(redirs) > 1:\n            lgr.warning(\"Multiple redirects aren't supported yet.  Taking first\")\n        elif len(redirs) == 0:\n            raise AccessFailedError(\"No redirects were resolved\")\n        raise UnhandledRedirectError(err_msg, url=redirs[0].url)\n    else:\n        raise AccessFailedError(err_msg)\n\n\n@auto_repr\nclass HTTPBaseAuthenticator(Authenticator):\n    \"\"\"Base class for html_form and http_auth authenticators\n    \"\"\"\n    def __init__(self, url=None, failure_re=None, success_re=None,\n                 session_cookies=None, **kwargs):\n        \"\"\"\n        Parameters\n        ----------\n        url : str, optional\n          URL where to find the form/login to authenticate.  If not provided, an original query url\n          which will be provided to the __call__ of the authenticator will be used\n        failure_re : str or list of str, optional\n        success_re : str or list of str, optional\n          Regular expressions to determine if login has failed or succeeded.\n          TODO: we might condition when it gets ran\n        session_cookies : str or list of str, optional\n          Session cookies to store (besides auth response cookies)\n        \"\"\"\n        super(HTTPBaseAuthenticator, self).__init__(**kwargs)\n        self.url = url\n        self.failure_re = assure_list_from_str(failure_re)\n        self.success_re = assure_list_from_str(success_re)\n        self.session_cookies = assure_list_from_str(session_cookies)\n\n    def authenticate(self, url, credential, session, update=False):\n        # we should use specified URL for this authentication first\n        lgr.info(\"http session: Authenticating into session for %s\", url)\n        post_url = self.url if self.url else url\n        credentials = credential()\n\n        # The whole thing relies on server first spitting out 401\n        # and client GETing again with 'Authentication:' header\n        # So we need custom handling for those, while keeping track not\n        # of cookies per se, but of 'Authentication:' header which is\n        # to be used in subsequent GETs\n        response = self._post_credential(credentials, post_url, session)\n        if response is None:\n            # authentication did not involve any interaction, nothing to\n            # check at this point\n            return\n\n        # Handle responses if there was initial authentication exchange,\n        # e.g. posting to a form and getting a cookie etc\n        err_prefix = \"Authentication to %s failed: \" % post_url\n        try:\n            check_response_status(response, err_prefix, session=session)\n        except DownloadError:\n            # It might have happened that the return code was 'incorrect'\n            # and we did get some feedback, which we could analyze to\n            # figure out actual problem.  E.g. in case of nersc of crcns\n            # it returns 404 (not found) with text in the html\n            if response is not None and response.text:\n                self.check_for_auth_failure(response.text, err_prefix)\n            raise\n\n        response_text = response.text\n        self.check_for_auth_failure(response_text, err_prefix)\n\n        if self.success_re:\n            # the one which must be used to verify success\n            # verify that we actually logged in\n            for success_re in self.success_re:\n                if not re.search(success_re, response_text):\n                    raise AccessDeniedError(\n                        err_prefix + \" returned output did not match 'success' regular expression %s\" % success_re\n                    )\n\n        cookies_dict = {}\n        if response.cookies:\n            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)\n        if self.session_cookies:\n            # any session cookies to store\n            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})\n\n        if cookies_dict:\n            if (url in cookies_db) and update:\n                cookies_db[url].update(cookies_dict)\n            else:\n                cookies_db[url] = cookies_dict\n            # assign cookies for this session\n            for c, v in cookies_dict.items():\n                if c not in session.cookies or session.cookies[c] != v:\n                    session.cookies[c] = v  # .update(cookies_dict)\n\n        return response\n\n    def _post_credential(self, credentials, post_url, session):\n        raise NotImplementedError(\"Must be implemented in subclass\")\n\n    def check_for_auth_failure(self, content, err_prefix=\"\"):\n        if self.failure_re:\n            # verify that we actually logged in\n            for failure_re in self.failure_re:\n                if re.search(failure_re, content):\n                    raise AccessDeniedError(\n                        err_prefix + \"returned output which matches regular expression %s\" % failure_re\n                    )\n\n\n@auto_repr\nclass HTMLFormAuthenticator(HTTPBaseAuthenticator):\n    \"\"\"Authenticate by opening a session via POSTing to HTML form\n    \"\"\"\n\n    def __init__(self, fields, tagid=None, **kwargs):\n        \"\"\"\n\n        Example specification in the .ini config file\n        [provider:crcns]\n        ...\n        credential = crcns ; is not given to authenticator as is\n        authentication_type = html_form\n        # TODO: may be rename into post_url\n        html_form_url = https://crcns.org/login_form\n        # probably not needed actually since form_url\n        # html_form_tagid = login_form\n        html_form_fields = __ac_name={user}\n                   __ac_password={password}\n                   submit=Log in\n                   form.submitted=1\n                   js_enabled=0\n                   cookies_enabled=\n        html_form_failure_re = (Login failed|Please log in)\n        html_form_success_re = You are now logged in\n\n        Parameters\n        ----------\n        fields : str or dict\n          String or a dictionary, which will be used (along with credential) information\n          to feed into the form\n        tagid : str, optional\n          id of the HTML <form> in the document to use. If None, and page contains a single form,\n          that one will be used.  If multiple forms -- error will be raise\n        **kwargs : dict, optional\n          Passed to super class HTTPBaseAuthenticator\n        \"\"\"\n        super(HTMLFormAuthenticator, self).__init__(**kwargs)\n        self.fields = assure_dict_from_str(fields)\n        self.tagid = tagid\n\n    def _post_credential(self, credentials, post_url, session):\n        post_fields = {\n            k: v.format(**credentials)\n            for k, v in self.fields.items()\n        }\n\n        response = session.post(post_url, data=post_fields)\n        lgr.debug(\"Posted to %s fields %s, got response %s with headers %s\",\n                  post_url, list(post_fields.keys()), response,\n                  list(response.headers.keys()))\n        return response\n\n\n@auto_repr\nclass HTTPRequestsAuthenticator(HTTPBaseAuthenticator):\n    \"\"\"Base class for various authenticators using requests pre-crafted ones\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = None\n    REQUESTS_FIELDS = ('user', 'password')\n\n    def __init__(self, **kwargs):\n        # so we have __init__ solely for a custom docstring\n        super(HTTPRequestsAuthenticator, self).__init__(**kwargs)\n\n    def _post_credential(self, credentials, post_url, session):\n        authenticator = self.REQUESTS_AUTHENTICATOR(\n            *[credentials[f] for f in self.REQUESTS_FIELDS])\n        session.auth = authenticator\n\n\n@auto_repr\nclass HTTPBasicAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via basic HTTP authentication\n\n    Example specification in the .ini config file\n    [provider:hcp-db]\n    ...\n    credential = hcp-db\n    authentication_type = http_auth\n\n    Parameters\n    ----------\n    **kwargs : dict, optional\n      Passed to super class HTTPBaseAuthenticator\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth\n\n\n@auto_repr\nclass HTTPAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via Basic authentication to some other post url\n\n    TODO:  actually this is some remnants which might later were RFed\n    into the form authenticator since otherwise they make little sense\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth\n\n    def _post_credential(self, credentials, post_url, session):\n        authenticator = self.REQUESTS_AUTHENTICATOR(\n            *[credentials[f] for f in self.REQUESTS_FIELDS])\n        session.auth = authenticator\n        response = session.post(post_url, data={},\n                                auth=authenticator)\n        return response\n\n\n@auto_repr\nclass HTTPDigestAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via HTTP digest authentication\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPDigestAuth\n\n\n@auto_repr\nclass HTTPBearerTokenAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via HTTP Authorization header\n    \"\"\"\n\n    DEFAULT_CREDENTIAL_TYPE = 'token'\n\n    def __init__(self, **kwargs):\n        # so we have __init__ solely for a custom docstring\n        super(HTTPBearerTokenAuthenticator, self).__init__(**kwargs)\n\n    def _post_credential(self, credentials, post_url, session):\n        # we do not need to post anything, just inject token into the session\n        session.headers['Authorization'] = \"Bearer %s\" % credentials['token']\n\n\n@auto_repr\nclass HTTPDownloaderSession(DownloaderSession):\n    def __init__(self, size=None, filename=None,  url=None, headers=None,\n                 response=None, chunk_size=1024 ** 2):\n        super(HTTPDownloaderSession, self).__init__(\n            size=size, filename=filename, url=url, headers=headers,\n        )\n        self.chunk_size = chunk_size\n        self.response = response\n\n    def download(self, f=None, pbar=None, size=None):\n        response = self.response\n        # content_gzipped = 'gzip' in response.headers.get('content-encoding', '').split(',')\n        # if content_gzipped:\n        #     raise NotImplemented(\"We do not support (yet) gzipped content\")\n        #     # see https://rationalpie.wordpress.com/2010/06/02/python-streaming-gzip-decompression/\n        #     # for ways to implement in python 2 and 3.2's gzip is working better with streams\n\n        total = 0\n        return_content = f is None\n        if f is None:\n            # no file to download to\n            # TODO: actually strange since it should have been decoded then...\n            f = io.BytesIO()\n\n        # must use .raw to be able avoiding decoding/decompression while downloading\n        # to a file\n        chunk_size_ = min(self.chunk_size, size) if size is not None else self.chunk_size\n\n        # XXX With requests_ftp BytesIO is provided as response.raw for ftp urls,\n        # which has no .stream, so let's do ducktyping and provide our custom stream\n        # via BufferedReader for such cases, while maintaining the rest of code\n        # intact.  TODO: figure it all out, since doesn't scale for any sizeable download\n        if not hasattr(response.raw, 'stream'):\n            def _stream():\n                buf = io.BufferedReader(response.raw)\n                v = True\n                while v:\n                    v = buf.read(chunk_size_)\n                    yield v\n\n            stream = _stream()\n        else:\n            # XXX TODO -- it must be just a dirty workaround\n            # As we discovered with downloads from NITRC all headers come with\n            # Content-Encoding: gzip which leads  requests to decode them.  But the point\n            # is that ftp links (yoh doesn't think) are gzip compressed for the transfer\n            decode_content = not response.url.startswith('ftp://')\n            stream = response.raw.stream(chunk_size_, decode_content=decode_content)\n\n        for chunk in stream:\n            if chunk:  # filter out keep-alive new chunks\n                chunk_len = len(chunk)\n                if size is not None and total + chunk_len > size:\n                    # trim the download to match target size\n                    chunk = chunk[:size - total]\n                    chunk_len = len(chunk)\n                total += chunk_len\n                f.write(chunk)\n                try:\n                    # TODO: pbar is not robust ATM against > 100% performance ;)\n                    if pbar:\n                        pbar.update(total)\n                except Exception as e:\n                    lgr.warning(\"Failed to update progressbar: %s\" % exc_str(e))\n                # TEMP\n                # see https://github.com/niltonvolpato/python-progressbar/pull/44\n                ui.out.flush()\n                if size is not None and total >= size:\n                    break  # we have done as much as we were asked\n\n        if return_content:\n            out = f.getvalue()\n            return out\n\n\n@auto_repr\nclass HTTPDownloader(BaseDownloader):\n    \"\"\"A stateful downloader to maintain a session to the website\n    \"\"\"\n\n    @borrowkwargs(BaseDownloader)\n    def __init__(self, headers={}, **kwargs):\n        super(HTTPDownloader, self).__init__(**kwargs)\n        self._session = None\n        self._headers = headers\n\n    def _establish_session(self, url, allow_old=True):\n        \"\"\"\n\n        Parameters\n        ----------\n        allow_old: bool, optional\n          If a Downloader allows for persistent sessions by some means -- flag\n          instructs whether to use previous session, or establish a new one\n\n        Returns\n        -------\n        bool\n          To state if old instance of a session/authentication was used\n        \"\"\"\n        if allow_old:\n            if self._session:\n                lgr.debug(\"http session: Reusing previous\")\n                return True  # we used old\n            elif url in cookies_db:\n                cookie_dict = cookies_db[url]\n                lgr.debug(\"http session: Creating new with old cookies %s\", list(cookie_dict.keys()))\n                self._session = requests.Session()\n                # not sure what happens if cookie is expired (need check to that or exception will prolly get thrown)\n\n                # TODO dict_to_cookiejar doesn't preserve all fields when reversed\n                self._session.cookies = requests.utils.cookiejar_from_dict(cookie_dict)\n                # TODO cookie could be expired w/ something like (but docs say it should be expired automatically):\n                # http://docs.python-requests.org/en/latest/api/#requests.cookies.RequestsCookieJar.clear_expired_cookies\n                # self._session.cookies.clear_expired_cookies()\n                return True\n\n        lgr.debug(\"http session: Creating brand new session\")\n        self._session = requests.Session()\n        if self.authenticator:\n            self.authenticator.authenticate(url, self.credential, self._session)\n\n        return False\n\n    def get_downloader_session(self, url,\n                               allow_redirects=True,\n                               use_redirected_url=True,\n                               headers=None):\n        # TODO: possibly make chunk size adaptive\n        # TODO: make it not this ugly -- but at the moment we are testing end-file size\n        # while can't know for sure if content was gunziped and whether it all went ok.\n        # So safer option -- just request to not have it gzipped\n        if headers is None:\n            headers = {}\n        if 'Accept-Encoding' not in headers:\n            headers['Accept-Encoding'] = ''\n        # TODO: our tests ATM aren't ready for retries, thus altogether disabled for now\n        nretries = 1\n        for retry in range(1, nretries+1):\n            try:\n                response = self._session.get(\n                    url, stream=True, allow_redirects=allow_redirects,\n                    headers=headers)\n            #except (MaxRetryError, NewConnectionError) as exc:\n            except Exception as exc:\n                # happen to run into those with urls pointing to Amazon,\n                # so let's rest and try again\n                if retry >= nretries:\n                    #import epdb; epdb.serve()\n                    raise AccessFailedError(\n                        \"Failed to establish a new session %d times. \"\n                        \"Last exception was: %s\"\n                        % (nretries, exc_str(exc)))\n                lgr.warning(\n                    \"Caught exception %s. Will retry %d out of %d times\",\n                    exc_str(exc), retry+1, nretries)\n                sleep(2**retry)\n\n        check_response_status(response, session=self._session)\n        headers = response.headers\n        lgr.debug(\"Establishing session for url %s, response headers: %s\",\n                  url, headers)\n        target_size = int(headers.get('Content-Length', '0').strip()) or None\n        if use_redirected_url and response.url and response.url != url:\n            lgr.debug(\"URL %s was redirected to %s and thus the later will be used\"\n                      % (url, response.url))\n            url = response.url\n        # Consult about filename.  Since we already have headers,\n        # should not result in an additional request\n        url_filename = get_url_filename(url, headers=headers)\n\n        headers['Url-Filename'] = url_filename\n        return HTTPDownloaderSession(\n            size=target_size,\n            url=response.url,\n            filename=url_filename,\n            headers=headers,\n            response=response\n        )\n\n    @classmethod\n    def get_status_from_headers(cls, headers):\n        \"\"\"Given HTTP headers, return 'status' record to assess later if link content was changed\n        \"\"\"\n        # used for quick checks for HTTP or S3?\n        # TODO:  So we will base all statuses on this set? e.g. for Last-Modified if to be\n        # mapping from field to its type converter\n        HTTP_HEADERS_TO_STATUS = {\n            'Content-Length': int,\n            'Content-Disposition': str,\n            'Last-Modified': rfc2822_to_epoch,\n            'Url-Filename': str,\n        }\n        # Allow for webserver to return them in other casing\n        HTTP_HEADERS_TO_STATUS_lower = {s.lower(): (s, t) for s, t in HTTP_HEADERS_TO_STATUS.items()}\n        status = {}\n        if headers:\n            for header_key in headers:\n                try:\n                    k, t = HTTP_HEADERS_TO_STATUS_lower[header_key.lower()]\n                except KeyError:\n                    continue\n                status[k] = t(headers[header_key])\n\n        # convert to FileStatus\n        return FileStatus(\n            size=status.get('Content-Length'),\n            mtime=status.get('Last-Modified'),\n            filename=get_response_disposition_filename(\n                status.get('Content-Disposition')) or status.get('Url-Filename')\n        )\n",
            "new_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Provide access to stuff (html, data files) via HTTP and HTTPS\n\n\"\"\"\nimport re\nimport requests\nimport requests.auth\n# at some point was trying to be too specific about which exceptions to\n# catch for a retry of a download.\n# from urllib3.exceptions import MaxRetryError, NewConnectionError\n\nimport io\nfrom time import sleep\n\nfrom ..utils import (\n    assure_list_from_str,\n    assure_dict_from_str,\n    ensure_bytes,\n)\nfrom ..dochelpers import borrowkwargs\n\nfrom ..ui import ui\nfrom ..utils import auto_repr\nfrom ..dochelpers import exc_str\nfrom ..support.network import get_url_filename\nfrom ..support.network import get_response_disposition_filename\nfrom ..support.network import rfc2822_to_epoch\nfrom ..support.cookies import cookies_db\nfrom ..support.status import FileStatus\nfrom ..support.exceptions import (\n    DownloadError,\n    AccessDeniedError,\n    AccessFailedError,\n    UnhandledRedirectError,\n)\n\nfrom .base import Authenticator\nfrom .base import BaseDownloader, DownloaderSession\n\nfrom logging import getLogger\nfrom ..log import LoggerHelper\nlgr = getLogger('datalad.http')\n\ntry:\n    import requests_ftp\n    requests_ftp.monkeypatch_session()\nexcept ImportError as e:\n    lgr.debug(\"Failed to import requests_ftp, thus no ftp support: %s\" % exc_str(e))\n    pass\n\nif lgr.getEffectiveLevel() <= 1:\n    # Let's also enable requests etc debugging\n\n    # These two lines enable debugging at httplib level (requests->urllib3->http.client)\n    # You will see the REQUEST, including HEADERS and DATA, and RESPONSE with HEADERS but without DATA.\n    # The only thing missing will be the response.body which is not logged.\n    import http.client\n    # TODO: nohow wrapped with logging, plain prints (heh heh), so formatting will not be consistent\n    http.client.HTTPConnection.debuglevel = 1\n\n    # for requests we can define logging properly\n    requests_log = LoggerHelper(logtarget=\"requests.packages.urllib3\").get_initialized_logger()\n    requests_log.setLevel(lgr.getEffectiveLevel())\n    requests_log.propagate = True\n\n__docformat__ = 'restructuredtext'\n\n\ndef process_www_authenticate(v):\n    if not v:\n        return []\n    supported_type = v.split(' ')[0].lower()\n    our_type = {\n        'basic': 'http_basic_auth',\n        'digest': 'http_digest_auth',\n    }.get(supported_type)\n    return [our_type] if our_type else []\n\n\ndef check_response_status(response, err_prefix=\"\", session=None):\n    \"\"\"Check if response's status_code signals problem with authentication etc\n\n    ATM succeeds only if response code was 200\n    \"\"\"\n    if not err_prefix:\n        err_prefix = \"Access to %s has failed: \" % response.url\n    # 401 would be for digest authentication mechanism, or if we first ask which mechanisms are\n    # supported.... must be linked into the logic if we decide to automagically detect which\n    # mechanism or to give more sensible error message\n    err_msg = err_prefix + \"status code %d\" % response.status_code\n    if response.status_code in {404}:\n        # It could have been that form_url is wrong, so let's just say that\n        # TODO: actually may be that is where we could use tagid and actually determine the form submission url\n        raise DownloadError(err_prefix + \"not found\")\n    elif 400 <= response.status_code < 500:\n        raise AccessDeniedError(\n            err_msg,\n            supported_types=process_www_authenticate(\n                response.headers.get('WWW-Authenticate')))\n    elif response.status_code in {200}:\n        pass\n    elif response.status_code in {301, 302, 307}:\n        # TODO: apparently tests do not excercise this one yet\n        if session is None:\n            raise AccessFailedError(err_msg + \" no session was provided\")\n        redirs = list(session.resolve_redirects(response, response.request))\n        if len(redirs) > 1:\n            lgr.warning(\"Multiple redirects aren't supported yet.  Taking first\")\n        elif len(redirs) == 0:\n            raise AccessFailedError(\"No redirects were resolved\")\n        raise UnhandledRedirectError(err_msg, url=redirs[0].url)\n    else:\n        raise AccessFailedError(err_msg)\n\n\n@auto_repr\nclass HTTPBaseAuthenticator(Authenticator):\n    \"\"\"Base class for html_form and http_auth authenticators\n    \"\"\"\n    def __init__(self, url=None, failure_re=None, success_re=None,\n                 session_cookies=None, **kwargs):\n        \"\"\"\n        Parameters\n        ----------\n        url : str, optional\n          URL where to find the form/login to authenticate.  If not provided, an original query url\n          which will be provided to the __call__ of the authenticator will be used\n        failure_re : str or list of str, optional\n        success_re : str or list of str, optional\n          Regular expressions to determine if login has failed or succeeded.\n          TODO: we might condition when it gets ran\n        session_cookies : str or list of str, optional\n          Session cookies to store (besides auth response cookies)\n        \"\"\"\n        super(HTTPBaseAuthenticator, self).__init__(**kwargs)\n        self.url = url\n        self.failure_re = assure_list_from_str(failure_re)\n        self.success_re = assure_list_from_str(success_re)\n        self.session_cookies = assure_list_from_str(session_cookies)\n\n    def authenticate(self, url, credential, session, update=False):\n        # we should use specified URL for this authentication first\n        lgr.info(\"http session: Authenticating into session for %s\", url)\n        post_url = self.url if self.url else url\n        credentials = credential()\n\n        # The whole thing relies on server first spitting out 401\n        # and client GETing again with 'Authentication:' header\n        # So we need custom handling for those, while keeping track not\n        # of cookies per se, but of 'Authentication:' header which is\n        # to be used in subsequent GETs\n        response = self._post_credential(credentials, post_url, session)\n        if response is None:\n            # authentication did not involve any interaction, nothing to\n            # check at this point\n            return\n\n        # Handle responses if there was initial authentication exchange,\n        # e.g. posting to a form and getting a cookie etc\n        err_prefix = \"Authentication to %s failed: \" % post_url\n        try:\n            check_response_status(response, err_prefix, session=session)\n        except DownloadError:\n            # It might have happened that the return code was 'incorrect'\n            # and we did get some feedback, which we could analyze to\n            # figure out actual problem.  E.g. in case of nersc of crcns\n            # it returns 404 (not found) with text in the html\n            if response is not None and response.text:\n                self.check_for_auth_failure(response.text, err_prefix)\n            raise\n\n        response_text = response.text\n        self.check_for_auth_failure(response_text, err_prefix)\n\n        if self.success_re:\n            # the one which must be used to verify success\n            # verify that we actually logged in\n            for success_re in self.success_re:\n                if not re.search(success_re, response_text):\n                    raise AccessDeniedError(\n                        err_prefix + \" returned output did not match 'success' regular expression %s\" % success_re\n                    )\n\n        cookies_dict = {}\n        if response.cookies:\n            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)\n        if self.session_cookies:\n            # any session cookies to store\n            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})\n\n        if cookies_dict:\n            if (url in cookies_db) and update:\n                cookies_db[url].update(cookies_dict)\n            else:\n                cookies_db[url] = cookies_dict\n            # assign cookies for this session\n            for c, v in cookies_dict.items():\n                if c not in session.cookies or session.cookies[c] != v:\n                    session.cookies[c] = v  # .update(cookies_dict)\n\n        return response\n\n    def _post_credential(self, credentials, post_url, session):\n        raise NotImplementedError(\"Must be implemented in subclass\")\n\n    def check_for_auth_failure(self, content, err_prefix=\"\"):\n        if self.failure_re:\n            content_is_bytes = isinstance(content, bytes)\n            # verify that we actually logged in\n            for failure_re in self.failure_re:\n                if content_is_bytes:\n                    failure_re = ensure_bytes(failure_re)\n                if re.search(failure_re, content):\n                    raise AccessDeniedError(\n                        err_prefix + \"returned output which matches regular expression %s\" % failure_re\n                    )\n\n\n@auto_repr\nclass HTMLFormAuthenticator(HTTPBaseAuthenticator):\n    \"\"\"Authenticate by opening a session via POSTing to HTML form\n    \"\"\"\n\n    def __init__(self, fields, tagid=None, **kwargs):\n        \"\"\"\n\n        Example specification in the .ini config file\n        [provider:crcns]\n        ...\n        credential = crcns ; is not given to authenticator as is\n        authentication_type = html_form\n        # TODO: may be rename into post_url\n        html_form_url = https://crcns.org/login_form\n        # probably not needed actually since form_url\n        # html_form_tagid = login_form\n        html_form_fields = __ac_name={user}\n                   __ac_password={password}\n                   submit=Log in\n                   form.submitted=1\n                   js_enabled=0\n                   cookies_enabled=\n        html_form_failure_re = (Login failed|Please log in)\n        html_form_success_re = You are now logged in\n\n        Parameters\n        ----------\n        fields : str or dict\n          String or a dictionary, which will be used (along with credential) information\n          to feed into the form\n        tagid : str, optional\n          id of the HTML <form> in the document to use. If None, and page contains a single form,\n          that one will be used.  If multiple forms -- error will be raise\n        **kwargs : dict, optional\n          Passed to super class HTTPBaseAuthenticator\n        \"\"\"\n        super(HTMLFormAuthenticator, self).__init__(**kwargs)\n        self.fields = assure_dict_from_str(fields)\n        self.tagid = tagid\n\n    def _post_credential(self, credentials, post_url, session):\n        post_fields = {\n            k: v.format(**credentials)\n            for k, v in self.fields.items()\n        }\n\n        response = session.post(post_url, data=post_fields)\n        lgr.debug(\"Posted to %s fields %s, got response %s with headers %s\",\n                  post_url, list(post_fields.keys()), response,\n                  list(response.headers.keys()))\n        return response\n\n\n@auto_repr\nclass HTTPRequestsAuthenticator(HTTPBaseAuthenticator):\n    \"\"\"Base class for various authenticators using requests pre-crafted ones\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = None\n    REQUESTS_FIELDS = ('user', 'password')\n\n    def __init__(self, **kwargs):\n        # so we have __init__ solely for a custom docstring\n        super(HTTPRequestsAuthenticator, self).__init__(**kwargs)\n\n    def _post_credential(self, credentials, post_url, session):\n        authenticator = self.REQUESTS_AUTHENTICATOR(\n            *[credentials[f] for f in self.REQUESTS_FIELDS])\n        session.auth = authenticator\n\n\n@auto_repr\nclass HTTPBasicAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via basic HTTP authentication\n\n    Example specification in the .ini config file\n    [provider:hcp-db]\n    ...\n    credential = hcp-db\n    authentication_type = http_auth\n\n    Parameters\n    ----------\n    **kwargs : dict, optional\n      Passed to super class HTTPBaseAuthenticator\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth\n\n\n@auto_repr\nclass HTTPAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via Basic authentication to some other post url\n\n    TODO:  actually this is some remnants which might later were RFed\n    into the form authenticator since otherwise they make little sense\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth\n\n    def _post_credential(self, credentials, post_url, session):\n        authenticator = self.REQUESTS_AUTHENTICATOR(\n            *[credentials[f] for f in self.REQUESTS_FIELDS])\n        session.auth = authenticator\n        response = session.post(post_url, data={},\n                                auth=authenticator)\n        return response\n\n\n@auto_repr\nclass HTTPDigestAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via HTTP digest authentication\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPDigestAuth\n\n\n@auto_repr\nclass HTTPBearerTokenAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via HTTP Authorization header\n    \"\"\"\n\n    DEFAULT_CREDENTIAL_TYPE = 'token'\n\n    def __init__(self, **kwargs):\n        # so we have __init__ solely for a custom docstring\n        super(HTTPBearerTokenAuthenticator, self).__init__(**kwargs)\n\n    def _post_credential(self, credentials, post_url, session):\n        # we do not need to post anything, just inject token into the session\n        session.headers['Authorization'] = \"Bearer %s\" % credentials['token']\n\n\n@auto_repr\nclass HTTPDownloaderSession(DownloaderSession):\n    def __init__(self, size=None, filename=None,  url=None, headers=None,\n                 response=None, chunk_size=1024 ** 2):\n        super(HTTPDownloaderSession, self).__init__(\n            size=size, filename=filename, url=url, headers=headers,\n        )\n        self.chunk_size = chunk_size\n        self.response = response\n\n    def download(self, f=None, pbar=None, size=None):\n        response = self.response\n        # content_gzipped = 'gzip' in response.headers.get('content-encoding', '').split(',')\n        # if content_gzipped:\n        #     raise NotImplemented(\"We do not support (yet) gzipped content\")\n        #     # see https://rationalpie.wordpress.com/2010/06/02/python-streaming-gzip-decompression/\n        #     # for ways to implement in python 2 and 3.2's gzip is working better with streams\n\n        total = 0\n        return_content = f is None\n        if f is None:\n            # no file to download to\n            # TODO: actually strange since it should have been decoded then...\n            f = io.BytesIO()\n\n        # must use .raw to be able avoiding decoding/decompression while downloading\n        # to a file\n        chunk_size_ = min(self.chunk_size, size) if size is not None else self.chunk_size\n\n        # XXX With requests_ftp BytesIO is provided as response.raw for ftp urls,\n        # which has no .stream, so let's do ducktyping and provide our custom stream\n        # via BufferedReader for such cases, while maintaining the rest of code\n        # intact.  TODO: figure it all out, since doesn't scale for any sizeable download\n        if not hasattr(response.raw, 'stream'):\n            def _stream():\n                buf = io.BufferedReader(response.raw)\n                v = True\n                while v:\n                    v = buf.read(chunk_size_)\n                    yield v\n\n            stream = _stream()\n        else:\n            # XXX TODO -- it must be just a dirty workaround\n            # As we discovered with downloads from NITRC all headers come with\n            # Content-Encoding: gzip which leads  requests to decode them.  But the point\n            # is that ftp links (yoh doesn't think) are gzip compressed for the transfer\n            decode_content = not response.url.startswith('ftp://')\n            stream = response.raw.stream(chunk_size_, decode_content=decode_content)\n\n        for chunk in stream:\n            if chunk:  # filter out keep-alive new chunks\n                chunk_len = len(chunk)\n                if size is not None and total + chunk_len > size:\n                    # trim the download to match target size\n                    chunk = chunk[:size - total]\n                    chunk_len = len(chunk)\n                total += chunk_len\n                f.write(chunk)\n                try:\n                    # TODO: pbar is not robust ATM against > 100% performance ;)\n                    if pbar:\n                        pbar.update(total)\n                except Exception as e:\n                    lgr.warning(\"Failed to update progressbar: %s\" % exc_str(e))\n                # TEMP\n                # see https://github.com/niltonvolpato/python-progressbar/pull/44\n                ui.out.flush()\n                if size is not None and total >= size:\n                    break  # we have done as much as we were asked\n\n        if return_content:\n            out = f.getvalue()\n            return out\n\n\n@auto_repr\nclass HTTPDownloader(BaseDownloader):\n    \"\"\"A stateful downloader to maintain a session to the website\n    \"\"\"\n\n    @borrowkwargs(BaseDownloader)\n    def __init__(self, headers={}, **kwargs):\n        super(HTTPDownloader, self).__init__(**kwargs)\n        self._session = None\n        self._headers = headers\n\n    def _establish_session(self, url, allow_old=True):\n        \"\"\"\n\n        Parameters\n        ----------\n        allow_old: bool, optional\n          If a Downloader allows for persistent sessions by some means -- flag\n          instructs whether to use previous session, or establish a new one\n\n        Returns\n        -------\n        bool\n          To state if old instance of a session/authentication was used\n        \"\"\"\n        if allow_old:\n            if self._session:\n                lgr.debug(\"http session: Reusing previous\")\n                return True  # we used old\n            elif url in cookies_db:\n                cookie_dict = cookies_db[url]\n                lgr.debug(\"http session: Creating new with old cookies %s\", list(cookie_dict.keys()))\n                self._session = requests.Session()\n                # not sure what happens if cookie is expired (need check to that or exception will prolly get thrown)\n\n                # TODO dict_to_cookiejar doesn't preserve all fields when reversed\n                self._session.cookies = requests.utils.cookiejar_from_dict(cookie_dict)\n                # TODO cookie could be expired w/ something like (but docs say it should be expired automatically):\n                # http://docs.python-requests.org/en/latest/api/#requests.cookies.RequestsCookieJar.clear_expired_cookies\n                # self._session.cookies.clear_expired_cookies()\n                return True\n\n        lgr.debug(\"http session: Creating brand new session\")\n        self._session = requests.Session()\n        if self.authenticator:\n            self.authenticator.authenticate(url, self.credential, self._session)\n\n        return False\n\n    def get_downloader_session(self, url,\n                               allow_redirects=True,\n                               use_redirected_url=True,\n                               headers=None):\n        # TODO: possibly make chunk size adaptive\n        # TODO: make it not this ugly -- but at the moment we are testing end-file size\n        # while can't know for sure if content was gunziped and whether it all went ok.\n        # So safer option -- just request to not have it gzipped\n        if headers is None:\n            headers = {}\n        if 'Accept-Encoding' not in headers:\n            headers['Accept-Encoding'] = ''\n        # TODO: our tests ATM aren't ready for retries, thus altogether disabled for now\n        nretries = 1\n        for retry in range(1, nretries+1):\n            try:\n                response = self._session.get(\n                    url, stream=True, allow_redirects=allow_redirects,\n                    headers=headers)\n            #except (MaxRetryError, NewConnectionError) as exc:\n            except Exception as exc:\n                # happen to run into those with urls pointing to Amazon,\n                # so let's rest and try again\n                if retry >= nretries:\n                    #import epdb; epdb.serve()\n                    raise AccessFailedError(\n                        \"Failed to establish a new session %d times. \"\n                        \"Last exception was: %s\"\n                        % (nretries, exc_str(exc)))\n                lgr.warning(\n                    \"Caught exception %s. Will retry %d out of %d times\",\n                    exc_str(exc), retry+1, nretries)\n                sleep(2**retry)\n\n        check_response_status(response, session=self._session)\n        headers = response.headers\n        lgr.debug(\"Establishing session for url %s, response headers: %s\",\n                  url, headers)\n        target_size = int(headers.get('Content-Length', '0').strip()) or None\n        if use_redirected_url and response.url and response.url != url:\n            lgr.debug(\"URL %s was redirected to %s and thus the later will be used\"\n                      % (url, response.url))\n            url = response.url\n        # Consult about filename.  Since we already have headers,\n        # should not result in an additional request\n        url_filename = get_url_filename(url, headers=headers)\n\n        headers['Url-Filename'] = url_filename\n        return HTTPDownloaderSession(\n            size=target_size,\n            url=response.url,\n            filename=url_filename,\n            headers=headers,\n            response=response\n        )\n\n    @classmethod\n    def get_status_from_headers(cls, headers):\n        \"\"\"Given HTTP headers, return 'status' record to assess later if link content was changed\n        \"\"\"\n        # used for quick checks for HTTP or S3?\n        # TODO:  So we will base all statuses on this set? e.g. for Last-Modified if to be\n        # mapping from field to its type converter\n        HTTP_HEADERS_TO_STATUS = {\n            'Content-Length': int,\n            'Content-Disposition': str,\n            'Last-Modified': rfc2822_to_epoch,\n            'Url-Filename': str,\n        }\n        # Allow for webserver to return them in other casing\n        HTTP_HEADERS_TO_STATUS_lower = {s.lower(): (s, t) for s, t in HTTP_HEADERS_TO_STATUS.items()}\n        status = {}\n        if headers:\n            for header_key in headers:\n                try:\n                    k, t = HTTP_HEADERS_TO_STATUS_lower[header_key.lower()]\n                except KeyError:\n                    continue\n                status[k] = t(headers[header_key])\n\n        # convert to FileStatus\n        return FileStatus(\n            size=status.get('Content-Length'),\n            mtime=status.get('Last-Modified'),\n            filename=get_response_disposition_filename(\n                status.get('Content-Disposition')) or status.get('Url-Filename')\n        )\n",
            "header": {
                "file": {
                    "path": "datalad/downloaders/http.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "a439ed542",
                "new_commit_hash": "297657001",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/downloaders/http.py"
            },
            "plus_file": {
                "path": "b/datalad/downloaders/http.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 19,
                            "length": 7
                        },
                        "new_range": {
                            "start": 19,
                            "length": 11
                        },
                        "section": "import requests.auth"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "import io",
                                "type": "context"
                            },
                            {
                                "content": "from time import sleep",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "from ..utils import assure_list_from_str, assure_dict_from_str",
                                "type": "deleted"
                            },
                            {
                                "content": "from ..utils import (",
                                "type": "added"
                            },
                            {
                                "content": "    assure_list_from_str,",
                                "type": "added"
                            },
                            {
                                "content": "    assure_dict_from_str,",
                                "type": "added"
                            },
                            {
                                "content": "    ensure_bytes,",
                                "type": "added"
                            },
                            {
                                "content": ")",
                                "type": "added"
                            },
                            {
                                "content": "from ..dochelpers import borrowkwargs",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "from ..ui import ui",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "datalad/downloaders/http.py",
                            "type": "statement",
                            "name": "",
                            "content": "from ..utils import (\n    assure_list_from_str,\n    assure_dict_from_str,\n    ensure_bytes,\n)",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 22,
                            "end_lineno": 26,
                            "parent": null
                        }
                    ],
                    "deleted_entities": [
                        {
                            "file_name": "datalad/downloaders/http.py",
                            "type": "statement",
                            "name": "",
                            "content": "from ..utils import assure_list_from_str, assure_dict_from_str",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 22,
                            "end_lineno": 22,
                            "parent": null
                        }
                    ]
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 208,
                            "length": 8
                        },
                        "new_range": {
                            "start": 212,
                            "length": 11
                        },
                        "section": "class HTTPBaseAuthenticator(Authenticator):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def check_for_auth_failure(self, content, err_prefix=\"\"):",
                                "type": "context"
                            },
                            {
                                "content": "        if self.failure_re:",
                                "type": "context"
                            },
                            {
                                "content": "            content_is_bytes = isinstance(content, bytes)",
                                "type": "added"
                            },
                            {
                                "content": "            # verify that we actually logged in",
                                "type": "context"
                            },
                            {
                                "content": "            for failure_re in self.failure_re:",
                                "type": "context"
                            },
                            {
                                "content": "                if content_is_bytes:",
                                "type": "added"
                            },
                            {
                                "content": "                    failure_re = ensure_bytes(failure_re)",
                                "type": "added"
                            },
                            {
                                "content": "                if re.search(failure_re, content):",
                                "type": "context"
                            },
                            {
                                "content": "                    raise AccessDeniedError(",
                                "type": "context"
                            },
                            {
                                "content": "                        err_prefix + \"returned output which matches regular expression %s\" % failure_re",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/downloaders/http.py",
                            "type": "class",
                            "name": "HTTPBaseAuthenticator",
                            "content": "class HTTPBaseAuthenticator(Authenticator):\n    \"\"\"Base class for html_form and http_auth authenticators\n    \"\"\"\n    def __init__(self, url=None, failure_re=None, success_re=None,\n                 session_cookies=None, **kwargs):\n        \"\"\"\n        Parameters\n        ----------\n        url : str, optional\n          URL where to find the form/login to authenticate.  If not provided, an original query url\n          which will be provided to the __call__ of the authenticator will be used\n        failure_re : str or list of str, optional\n        success_re : str or list of str, optional\n          Regular expressions to determine if login has failed or succeeded.\n          TODO: we might condition when it gets ran\n        session_cookies : str or list of str, optional\n          Session cookies to store (besides auth response cookies)\n        \"\"\"\n        super(HTTPBaseAuthenticator, self).__init__(**kwargs)\n        self.url = url\n        self.failure_re = assure_list_from_str(failure_re)\n        self.success_re = assure_list_from_str(success_re)\n        self.session_cookies = assure_list_from_str(session_cookies)\n\n    def authenticate(self, url, credential, session, update=False):\n        # we should use specified URL for this authentication first\n        lgr.info(\"http session: Authenticating into session for %s\", url)\n        post_url = self.url if self.url else url\n        credentials = credential()\n\n        # The whole thing relies on server first spitting out 401\n        # and client GETing again with 'Authentication:' header\n        # So we need custom handling for those, while keeping track not\n        # of cookies per se, but of 'Authentication:' header which is\n        # to be used in subsequent GETs\n        response = self._post_credential(credentials, post_url, session)\n        if response is None:\n            # authentication did not involve any interaction, nothing to\n            # check at this point\n            return\n\n        # Handle responses if there was initial authentication exchange,\n        # e.g. posting to a form and getting a cookie etc\n        err_prefix = \"Authentication to %s failed: \" % post_url\n        try:\n            check_response_status(response, err_prefix, session=session)\n        except DownloadError:\n            # It might have happened that the return code was 'incorrect'\n            # and we did get some feedback, which we could analyze to\n            # figure out actual problem.  E.g. in case of nersc of crcns\n            # it returns 404 (not found) with text in the html\n            if response is not None and response.text:\n                self.check_for_auth_failure(response.text, err_prefix)\n            raise\n\n        response_text = response.text\n        self.check_for_auth_failure(response_text, err_prefix)\n\n        if self.success_re:\n            # the one which must be used to verify success\n            # verify that we actually logged in\n            for success_re in self.success_re:\n                if not re.search(success_re, response_text):\n                    raise AccessDeniedError(\n                        err_prefix + \" returned output did not match 'success' regular expression %s\" % success_re\n                    )\n\n        cookies_dict = {}\n        if response.cookies:\n            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)\n        if self.session_cookies:\n            # any session cookies to store\n            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})\n\n        if cookies_dict:\n            if (url in cookies_db) and update:\n                cookies_db[url].update(cookies_dict)\n            else:\n                cookies_db[url] = cookies_dict\n            # assign cookies for this session\n            for c, v in cookies_dict.items():\n                if c not in session.cookies or session.cookies[c] != v:\n                    session.cookies[c] = v  # .update(cookies_dict)\n\n        return response\n\n    def _post_credential(self, credentials, post_url, session):\n        raise NotImplementedError(\"Must be implemented in subclass\")\n\n    def check_for_auth_failure(self, content, err_prefix=\"\"):\n        if self.failure_re:\n            content_is_bytes = isinstance(content, bytes)\n            # verify that we actually logged in\n            for failure_re in self.failure_re:\n                if content_is_bytes:\n                    failure_re = ensure_bytes(failure_re)\n                if re.search(failure_re, content):\n                    raise AccessDeniedError(\n                        err_prefix + \"returned output which matches regular expression %s\" % failure_re\n                    )",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 124,
                            "end_lineno": 223,
                            "parent": null
                        },
                        {
                            "file_name": "datalad/downloaders/http.py",
                            "type": "method",
                            "name": "HTTPBaseAuthenticator.check_for_auth_failure",
                            "content": "class HTTPBaseAuthenticator(Authenticator):\n    \"\"\"Base class for html_form and http_auth authenticators\n    \"\"\"\n    def __init__(self, url=None, failure_re=None, success_re=None,\n                 session_cookies=None, **kwargs):\n        \"\"\"\n        Parameters\n        ----------\n        url : str, optional\n          URL where to find the form/login to authenticate.  If not provided, an original query url\n          which will be provided to the __call__ of the authenticator will be used\n        failure_re : str or list of str, optional\n        success_re : str or list of str, optional\n          Regular expressions to determine if login has failed or succeeded.\n          TODO: we might condition when it gets ran\n        session_cookies : str or list of str, optional\n          Session cookies to store (besides auth response cookies)\n        \"\"\"\n        super(HTTPBaseAuthenticator, self).__init__(**kwargs)\n        self.url = url\n        self.failure_re = assure_list_from_str(failure_re)\n        self.success_re = assure_list_from_str(success_re)\n        self.session_cookies = assure_list_from_str(session_cookies)\n\n    def authenticate(self, url, credential, session, update=False):\n        # we should use specified URL for this authentication first\n        lgr.info(\"http session: Authenticating into session for %s\", url)\n        post_url = self.url if self.url else url\n        credentials = credential()\n\n        # The whole thing relies on server first spitting out 401\n        # and client GETing again with 'Authentication:' header\n        # So we need custom handling for those, while keeping track not\n        # of cookies per se, but of 'Authentication:' header which is\n        # to be used in subsequent GETs\n        response = self._post_credential(credentials, post_url, session)\n        if response is None:\n            # authentication did not involve any interaction, nothing to\n            # check at this point\n            return\n\n        # Handle responses if there was initial authentication exchange,\n        # e.g. posting to a form and getting a cookie etc\n        err_prefix = \"Authentication to %s failed: \" % post_url\n        try:\n            check_response_status(response, err_prefix, session=session)\n        except DownloadError:\n            # It might have happened that the return code was 'incorrect'\n            # and we did get some feedback, which we could analyze to\n            # figure out actual problem.  E.g. in case of nersc of crcns\n            # it returns 404 (not found) with text in the html\n            if response is not None and response.text:\n                self.check_for_auth_failure(response.text, err_prefix)\n            raise\n\n        response_text = response.text\n        self.check_for_auth_failure(response_text, err_prefix)\n\n        if self.success_re:\n            # the one which must be used to verify success\n            # verify that we actually logged in\n            for success_re in self.success_re:\n                if not re.search(success_re, response_text):\n                    raise AccessDeniedError(\n                        err_prefix + \" returned output did not match 'success' regular expression %s\" % success_re\n                    )\n\n        cookies_dict = {}\n        if response.cookies:\n            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)\n        if self.session_cookies:\n            # any session cookies to store\n            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})\n\n        if cookies_dict:\n            if (url in cookies_db) and update:\n                cookies_db[url].update(cookies_dict)\n            else:\n                cookies_db[url] = cookies_dict\n            # assign cookies for this session\n            for c, v in cookies_dict.items():\n                if c not in session.cookies or session.cookies[c] != v:\n                    session.cookies[c] = v  # .update(cookies_dict)\n\n        return response\n\n    def _post_credential(self, credentials, post_url, session):\n        raise NotImplementedError(\"Must be implemented in subclass\")\n\n    def check_for_auth_failure(self, content, err_prefix=\"\"):\n        if self.failure_re:\n            content_is_bytes = isinstance(content, bytes)\n            # verify that we actually logged in\n            for failure_re in self.failure_re:\n                if content_is_bytes:\n                    failure_re = ensure_bytes(failure_re)\n                if re.search(failure_re, content):\n                    raise AccessDeniedError(\n                        err_prefix + \"returned output which matches regular expression %s\" % failure_re\n                    )",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 213,
                            "end_lineno": 223,
                            "parent": {
                                "file_name": "datalad/downloaders/http.py",
                                "type": "class",
                                "name": "HTTPBaseAuthenticator",
                                "content": "class HTTPBaseAuthenticator(Authenticator):\n    \"\"\"Base class for html_form and http_auth authenticators\n    \"\"\"\n    def __init__(self, url=None, failure_re=None, success_re=None,\n                 session_cookies=None, **kwargs):\n        \"\"\"\n        Parameters\n        ----------\n        url : str, optional\n          URL where to find the form/login to authenticate.  If not provided, an original query url\n          which will be provided to the __call__ of the authenticator will be used\n        failure_re : str or list of str, optional\n        success_re : str or list of str, optional\n          Regular expressions to determine if login has failed or succeeded.\n          TODO: we might condition when it gets ran\n        session_cookies : str or list of str, optional\n          Session cookies to store (besides auth response cookies)\n        \"\"\"\n        super(HTTPBaseAuthenticator, self).__init__(**kwargs)\n        self.url = url\n        self.failure_re = assure_list_from_str(failure_re)\n        self.success_re = assure_list_from_str(success_re)\n        self.session_cookies = assure_list_from_str(session_cookies)\n\n    def authenticate(self, url, credential, session, update=False):\n        # we should use specified URL for this authentication first\n        lgr.info(\"http session: Authenticating into session for %s\", url)\n        post_url = self.url if self.url else url\n        credentials = credential()\n\n        # The whole thing relies on server first spitting out 401\n        # and client GETing again with 'Authentication:' header\n        # So we need custom handling for those, while keeping track not\n        # of cookies per se, but of 'Authentication:' header which is\n        # to be used in subsequent GETs\n        response = self._post_credential(credentials, post_url, session)\n        if response is None:\n            # authentication did not involve any interaction, nothing to\n            # check at this point\n            return\n\n        # Handle responses if there was initial authentication exchange,\n        # e.g. posting to a form and getting a cookie etc\n        err_prefix = \"Authentication to %s failed: \" % post_url\n        try:\n            check_response_status(response, err_prefix, session=session)\n        except DownloadError:\n            # It might have happened that the return code was 'incorrect'\n            # and we did get some feedback, which we could analyze to\n            # figure out actual problem.  E.g. in case of nersc of crcns\n            # it returns 404 (not found) with text in the html\n            if response is not None and response.text:\n                self.check_for_auth_failure(response.text, err_prefix)\n            raise\n\n        response_text = response.text\n        self.check_for_auth_failure(response_text, err_prefix)\n\n        if self.success_re:\n            # the one which must be used to verify success\n            # verify that we actually logged in\n            for success_re in self.success_re:\n                if not re.search(success_re, response_text):\n                    raise AccessDeniedError(\n                        err_prefix + \" returned output did not match 'success' regular expression %s\" % success_re\n                    )\n\n        cookies_dict = {}\n        if response.cookies:\n            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)\n        if self.session_cookies:\n            # any session cookies to store\n            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})\n\n        if cookies_dict:\n            if (url in cookies_db) and update:\n                cookies_db[url].update(cookies_dict)\n            else:\n                cookies_db[url] = cookies_dict\n            # assign cookies for this session\n            for c, v in cookies_dict.items():\n                if c not in session.cookies or session.cookies[c] != v:\n                    session.cookies[c] = v  # .update(cookies_dict)\n\n        return response\n\n    def _post_credential(self, credentials, post_url, session):\n        raise NotImplementedError(\"Must be implemented in subclass\")\n\n    def check_for_auth_failure(self, content, err_prefix=\"\"):\n        if self.failure_re:\n            content_is_bytes = isinstance(content, bytes)\n            # verify that we actually logged in\n            for failure_re in self.failure_re:\n                if content_is_bytes:\n                    failure_re = ensure_bytes(failure_re)\n                if re.search(failure_re, content):\n                    raise AccessDeniedError(\n                        err_prefix + \"returned output which matches regular expression %s\" % failure_re\n                    )",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 124,
                                "end_lineno": 223,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Tests for http downloader\"\"\"\n\nimport time\nfrom calendar import timegm\nimport re\n\nimport os\nimport builtins\nfrom os.path import join as opj\n\nfrom datalad.downloaders.tests.utils import get_test_providers\nfrom ..base import (\n    BaseDownloader,\n    DownloadError,\n    IncompleteDownloadError,\n    NoneAuthenticator,\n)\nfrom ..credentials import (\n    LORIS_Token,\n    Token,\n    UserPassword,\n)\nfrom ..http import (\n    HTMLFormAuthenticator,\n    HTTPDownloader,\n    HTTPBearerTokenAuthenticator,\n    process_www_authenticate,\n)\nfrom ...support.exceptions import AccessFailedError\nfrom ...support.network import get_url_straight_filename\nfrom ...utils import ensure_unicode\n\n# BTW -- mock_open is not in mock on wheezy (Debian 7.x)\ntry:\n    import httpretty\nexcept (ImportError, AttributeError):\n    # Attribute Error happens with newer httpretty and older ssl module\n    # https://github.com/datalad/datalad/pull/2623\n    class NoHTTPPretty(object):\n       __bool__ = lambda s: False\n       activate = lambda s, t: t\n    httpretty = NoHTTPPretty()\n\nfrom unittest.mock import patch\nfrom ...tests.utils import (\n    SkipTest,\n    assert_equal,\n    assert_false,\n    assert_greater,\n    assert_in,\n    assert_not_in,\n    assert_raises,\n    known_failure_githubci_win,\n    ok_file_has_content,\n    serve_path_via_http, with_tree,\n    skip_if,\n    skip_if_no_network,\n    swallow_logs,\n    swallow_outputs,\n    use_cassette,\n    with_fake_cookies_db,\n    with_memory_keyring,\n    with_tempfile,\n    with_testsui,\n    without_http_proxy,\n)\nfrom ...support.exceptions import (\n    AccessDeniedError,\n    AnonymousAccessDeniedError,\n)\nfrom ...support.status import FileStatus\nfrom ...support.network import get_url_disposition_filename\n\n\ndef test_docstring():\n    doc = HTTPDownloader.__init__.__doc__\n    assert_in(\"\\ncredential: Credential\", doc)\n\n\n# XXX doesn't quite work as it should since doesn't provide context handling\n# I guess... but at least causes the DownloadError ;)\n_builtins_open = builtins.open\n\n\ndef fake_open(write_=None, skip_regex=None):\n    class myfile(object):\n        \"\"\"file which does nothing\"\"\"\n        if write_:\n            def write(self, *args, **kwargs):\n                write_(*args, **kwargs)\n        def close(self):\n            pass\n\n    def myopen(path, *args, **kwargs):\n        if skip_regex and re.search(skip_regex, ensure_unicode(path)):\n            return _builtins_open(path, *args, **kwargs)\n        else:\n            return myfile\n    return myopen\n\n\ndef _raise_IOError(*args, **kwargs):\n    raise IOError(\"Testing here\")\n\n\ndef test_process_www_authenticate():\n    assert_equal(process_www_authenticate(\"Basic\"),\n                 [\"http_basic_auth\"])\n    assert_equal(process_www_authenticate(\"Digest\"),\n                 [\"http_digest_auth\"])\n    assert_equal(process_www_authenticate(\"Digest more\"),\n                 [\"http_digest_auth\"])\n    assert_equal(process_www_authenticate(\"Unknown\"),\n                 [])\n\n\n@known_failure_githubci_win\n@with_tree(tree=[('file.dat', 'abc')])\n@serve_path_via_http\ndef test_HTTPDownloader_basic(toppath, topurl):\n    furl = \"%sfile.dat\" % topurl\n    tfpath = opj(toppath, \"file-downloaded.dat\")\n    downloader = HTTPDownloader()  # no auth/credentials needed\n    download = downloader.download\n    download(furl, tfpath)\n    ok_file_has_content(tfpath, 'abc')\n\n    # download() creates leading directories if needed for file targets...\n    subdir_tfpath = opj(toppath, \"l1\", \"l2\", \"file-downloaded.dat\")\n    download(furl, subdir_tfpath)\n    ok_file_has_content(subdir_tfpath, 'abc')\n\n    # ... and for directory targets.\n    subdir_dirtarget = opj(toppath, \"d1\", \"d2\", \"\")\n    download(furl, subdir_dirtarget)\n    ok_file_has_content(opj(subdir_dirtarget, \"file.dat\"), \"abc\")\n\n    # see if fetch works correctly\n    assert_equal(downloader.fetch(furl), 'abc')\n\n    # By default should not overwrite the file\n    assert_raises(DownloadError, download, furl, tfpath)\n    # but be able to redownload whenever overwrite==True\n    downloaded_path = download(furl, tfpath, overwrite=True)\n    assert_equal(downloaded_path, tfpath)\n    ok_file_has_content(tfpath, 'abc')\n\n    # Fail with an informative message if we're downloading into a directory\n    # and the file name can't be determined from the URL.\n    with assert_raises(DownloadError) as cm:\n        download(topurl, toppath)\n    assert_in(\"File name could not be determined\", str(cm.exception))\n\n    # Some errors handling\n    # XXX obscure mocking since impossible to mock write alone\n    # and it still results in some warning being spit out\n    # Note: we need to avoid mocking opening of the lock file!\n    with swallow_logs(), \\\n         patch.object(builtins, 'open', fake_open(write_=_raise_IOError, skip_regex='.*\\.lck$')):\n        assert_raises(DownloadError, download, furl, tfpath, overwrite=True)\n\n    # incomplete download scenario - should have 3 tries\n    def _fail_verify_download(try_to_fail):\n        try_ = [0]\n        _orig_verify_download = BaseDownloader._verify_download\n        def _verify_download(self, *args, **kwargs):\n            try_[0] += 1\n            if try_[0] >= try_to_fail:\n                return _orig_verify_download(self, *args, **kwargs)\n            raise IncompleteDownloadError()\n        return _verify_download\n\n    with patch.object(BaseDownloader, '_verify_download', _fail_verify_download(6)), \\\n        swallow_logs():\n            # how was before the \"fix\":\n            #assert_raises(DownloadError, downloader.fetch, furl)\n            #assert_raises(DownloadError, downloader.fetch, furl)\n            # now should download just fine\n            assert_equal(downloader.fetch(furl), 'abc')\n    # but should fail if keeps failing all 5 times and then on 11th should raise DownloadError\n    with patch.object(BaseDownloader, '_verify_download', _fail_verify_download(7)), \\\n        swallow_logs():\n            assert_raises(DownloadError, downloader.fetch, furl)\n\n    # TODO: access denied scenario\n    # TODO: access denied detection\n\n\n@with_tree(tree=[('file.dat', 'abc')])\n@serve_path_via_http\n@with_memory_keyring\ndef test_access_denied(toppath, topurl, keyring):\n    furl = topurl + \"file.dat\"\n\n    def deny_access(*args, **kwargs):\n        raise AccessDeniedError(supported_types=[\"http_basic_auth\"])\n\n    def deny_anon_access(*args, **kwargs):\n        raise AnonymousAccessDeniedError(supported_types=[\"http_basic_auth\"])\n\n    downloader = HTTPDownloader()\n\n    # Test different paths that should lead to a DownloadError.\n\n    for denier in deny_access, deny_anon_access:\n        @with_testsui(responses=[\"no\"])\n        def run_refuse_provider_setup():\n            with patch.object(downloader, '_download', denier):\n                downloader.download(furl)\n        assert_raises(DownloadError, run_refuse_provider_setup)\n\n    downloader_creds = HTTPDownloader(credential=\"irrelevant\")\n\n    @with_testsui(responses=[\"no\"])\n    def run_refuse_creds_update():\n        with patch.object(downloader_creds, '_download', deny_access):\n            downloader_creds.download(furl)\n    assert_raises(DownloadError, run_refuse_creds_update)\n\n    downloader_noauth = HTTPDownloader(authenticator=NoneAuthenticator())\n\n    def run_noauth():\n        with patch.object(downloader_noauth, '_download', deny_access):\n            downloader_noauth.download(furl)\n    assert_raises(DownloadError, run_noauth)\n\n    # Complete setup for a new provider.\n\n    @with_testsui(responses=[\n        \"yes\",  # Set up provider?\n        # Enter provider details, but then don't save ...\n        \"newprovider\", re.escape(furl), \"http_auth\", \"user_password\", \"no\",\n        # No provider, try again?\n        \"yes\",\n        # Enter same provider detains but save this time.\n        \"newprovider\", re.escape(furl), \"http_auth\", \"user_password\", \"yes\",\n        # Enter credentials.\n        \"me\", \"mypass\"\n    ])\n    def run_set_up_provider():\n        with patch.object(downloader, '_download', deny_access):\n            downloader.download(furl)\n\n    # We've forced an AccessDenied error and then set up bogus credentials,\n    # leading to a 501 (not implemented) error.\n    assert_raises(AccessFailedError, run_set_up_provider)\n\n\n@with_tempfile(mkdir=True)\ndef check_download_external_url(url, failed_str, success_str, d, url_final=None):\n    fpath = opj(d, get_url_straight_filename(url))\n    providers = get_test_providers(url)  # url for check of credentials\n    provider = providers.get_provider(url)\n    downloader = provider.get_downloader(url)\n\n    # Download way\n    with swallow_outputs() as cmo:\n        downloaded_path = downloader.download(url, path=d)\n    assert_equal(fpath, downloaded_path)\n    with open(fpath) as f:\n        content = f.read()\n        if success_str is not None:\n            assert_in(success_str, content)\n        if failed_str is not None:\n            assert_false(failed_str in content)\n\n    # And if we specify size\n    for s in [1, 2]:\n        with swallow_outputs() as cmo:\n            downloaded_path_ = downloader.download(url, path=d, size=s, overwrite=True)\n        # should not be affected\n        assert_equal(downloaded_path, downloaded_path_)\n        with open(fpath) as f:\n            content_ = f.read()\n        assert_equal(len(content_), s)\n        assert_equal(content_, content[:s])\n\n    # Fetch way\n    content = downloader.fetch(url)\n    if success_str is not None:\n        assert_in(success_str, content)\n    if failed_str is not None:\n        assert_false(failed_str in content)\n\n    # And if we specify size\n    for s in [1, 2]:\n        with swallow_outputs() as cmo:\n            content_ = downloader.fetch(url, size=s)\n        assert_equal(len(content_), s)\n        assert_equal(content_, content[:s])\n\n    # Verify status\n    status = downloader.get_status(url)\n    assert(isinstance(status, FileStatus))\n    if not url.startswith('ftp://'):\n        # TODO introduce support for mtime into requests_ftp?\n        assert(status.mtime)\n    assert(status.size)\n\n    # Verify possible redirections\n    if url_final is None:\n        url_final = url\n    assert_equal(downloader.get_target_url(url), url_final)\n\n    # TODO -- more and more specific\n\n\n# TODO: @use_cassette is not playing nice with generators, causing\n# troubles when trying to cause test skip if no network. So disabling for now\n# https://github.com/datalad/datalad/issues/3158\n# @use_cassette('test_authenticate_external_portals', record_mode='once')\ndef test_authenticate_external_portals():\n    skip_if_no_network()\n    yield check_download_external_url, \\\n          \"https://portal.nersc.gov/project/crcns/download/alm-1/checksums.md5\", \\\n          \"<form action=\", \\\n          \"datafiles/meta_data_files.tar.gz\"\n    # seems to be gone\n    # yield check_download_external_url, \\\n    #       'https://db.humanconnectome.org/data/archive/projects/HCP_500/subjects/100307/experiments/100307_CREST/resources/100307_CREST/files/unprocessed/3T/Diffusion/100307_3T_DWI_dir97_LR.bval', \\\n    #       \"failed\", \\\n    #       \"2000 1005 2000 3000\"\n    yield check_download_external_url, \\\n          'https://db.humanconnectome.org/data/experiments/ConnectomeDB_E09797/resources/166768/files/filescans.csv', \\\n          \"failed\", \\\n          \"'Scan','FilePath'\"\ntest_authenticate_external_portals.tags = ['external-portal', 'network']\n\n\n@known_failure_githubci_win\n@skip_if_no_network\ndef test_download_ftp():\n    try:\n        import requests_ftp\n    except ImportError:\n        raise SkipTest(\"need requests_ftp\")  # TODO - make it not ad-hoc\n    try:\n        check_download_external_url(\n                  \"ftp://ftp.gnu.org/README\",\n                  None,\n                  \"This is ftp.gnu.org\"\n        )\n    except AccessFailedError as exc:\n        if 'status code 503' in str(exc):\n            raise SkipTest(\"ftp.gnu.org throws 503 when on travis (only?)\")\n        raise\n\n\n# TODO: redo smart way with mocking, to avoid unnecessary CPU waste\n@with_tree(tree={'file.dat': '1'})\n@serve_path_via_http\n@with_tempfile\ndef test_mtime(path, url, tempfile):\n    # let's set custom mtime\n    file_to_download = opj(path, 'file.dat')\n    os.utime(file_to_download, (time.time(), 1000))\n    assert_equal(os.stat(file_to_download).st_mtime, 1000)\n\n    file_url = \"%s/%s\" % (url, 'file.dat')\n    with swallow_outputs():\n        get_test_providers().download(file_url, path=tempfile)\n    assert_equal(os.stat(tempfile).st_mtime, 1000)\n\n\ndef test_get_status_from_headers():\n    # function doesn't do any value transformation ATM\n    headers = {\n        'Content-Length': '123',\n        # some other file record - we don't test content here yet\n        'Content-Disposition': 'attachment; filename=\"bogus.txt\"',\n        'Last-Modified': 'Sat, 07 Nov 2015 05:23:36 GMT'\n    }\n    headers['bogus1'] = '123'\n\n    assert_equal(\n            HTTPDownloader.get_status_from_headers(headers),\n            FileStatus(size=123, filename='bogus.txt', mtime=1446873816))\n\n    assert_equal(HTTPDownloader.get_status_from_headers({'content-lengtH': '123'}),\n                 FileStatus(size=123))\n\n    filename = 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'\n    headers_content_disposition = {\n        'Content-Disposition':\n            'Attachment;Filename=\"%s\"' % filename, }\n    assert_equal(\n        HTTPDownloader.get_status_from_headers(headers_content_disposition).filename,\n        filename)\n\n    # since we are providing full headers -- irrelevant\n    assert_equal(get_url_disposition_filename(\"http://irrelevant\", headers_content_disposition),\n                 filename)\n\n\n\n# TODO: test that download fails (even if authentication credentials are right) if form_url\n# is wrong!\n\n\nclass FakeCredential1(UserPassword):\n    \"\"\"Credential to test scenarios.\"\"\"\n    # to be reusable, and not leak across tests,\n    # we should get _fixed_credentials per instance\n    def __init__(self, *args, **kwargs):\n        super(FakeCredential1, self).__init__(*args, **kwargs)\n        self._fixed_credentials = [\n            {'user': 'testlogin', 'password': 'testpassword'},\n            {'user': 'testlogin2', 'password': 'testpassword2'},\n            {'user': 'testlogin2', 'password': 'testpassword3'}\n        ]\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials[0]\n    def enter_new(self):\n        # pop last used credential, so we would use \"new\" ones\n        del self._fixed_credentials[0]\n\n\nurl = \"http://example.com/crap.txt\"\ntest_cookie = 'somewebsite=testcookie'\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTMLFormAuthenticator_httpretty(d):\n    fpath = opj(d, 'crap.txt')\n\n    credential = FakeCredential1(name='test', url=None)\n    credentials = credential()\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        assert_equal(credentials['password'], post_params['password'][0])\n        assert_equal(credentials['user'], post_params['username'][0])\n        assert_not_in('Cookie', request.headers)\n        return (200, headers, \"Got {} response from {}\".format(request.method, uri))\n\n    def request_get_callback(request, uri, headers):\n        assert_equal(request.body, b'')\n        assert_in('Cookie', request.headers)\n        assert_equal(request.headers['Cookie'], test_cookie)\n        return (200, headers, \"correct body\")\n\n    # SCENARIO 1\n    # callback to verify that correct credentials are provided\n    # and then returns the cookie to test again for 'GET'ing\n    httpretty.register_uri(httpretty.POST, url,\n                           body=request_post_callback,\n                           set_cookie=test_cookie)\n    # then in GET verify that correct cookie was provided and\n    # that no credentials are there\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n    # SCENARIO 2\n    # outdated cookie provided to GET -- must return 403 (access denied)\n    # then our code should POST credentials again and get a new cookies\n    # which is then provided to GET\n\n    # SCENARIO 3\n    # outdated cookie\n    # outdated credentials\n    # it should ask for new credentials (FakeCredential1 already mocks for that)\n    # and then SCENARIO1 must work again\n\n    # SCENARIO 4\n    # cookie and credentials expired, user provided new bad credential\n\n    # Also we want to test how would it work if cookie is available (may be)\n    # TODO: check with correct and incorrect credential\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"))\n    # TODO: with success_re etc\n    # This is a \"success test\" which should be tested in various above scenarios\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n    # Unsuccesfull scenarios to test:\n    # the provided URL at the end 404s, or another failure (e.g. interrupted download)\n\n\n@with_memory_keyring\n@with_testsui(responses=['no', 'yes', 'testlogin', 'testpassword'])\ndef test_auth_but_no_cred(keyring):\n    authenticator = HTMLFormAuthenticator(\"\")\n    # Replying 'no' to the set credentials prompt should raise ValueError\n    assert_raises(ValueError, HTTPDownloader, credential=None, authenticator=authenticator)\n    # Reply 'yes' and set test user:pass at the next set credentials prompt\n    downloader = HTTPDownloader(credential=None, authenticator=authenticator)\n    # Verify credentials correctly set to test user:pass\n    assert_equal(downloader.credential.get('user'), 'testlogin')\n    assert_equal(downloader.credential.get('password'), 'testpassword')\n\n\n@with_testsui(responses=['yes'])  # will request to reentry it\ndef test_authfail404_interactive():\n    # we will firsts get 'failed' but then real 404 when trying new password\n    check_httpretty_authfail404(['failed', '404'])\n\n\n@with_testsui(interactive=False)  # no interactions -- blow!\ndef test_authfail404_noninteractive():\n    # we do not get to the 2nd attempt so just get 'failed'\n    # and exception thrown inside is not emerging all the way here but\n    # caught in the check_\n    check_httpretty_authfail404(['failed'])\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_fake_cookies_db\n@with_tempfile(mkdir=True)\ndef check_httpretty_authfail404(exp_called, d):\n    # mimic behavior of nersc which 404s but provides feedback whenever\n    # credentials are incorrect.  In our case we should fail properly\n    credential = FakeCredential1(name='test', url=None)\n\n    was_called = []\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        if post_params['password'][0] == 'testpassword2':\n            was_called.append('404')\n            return 404, headers, \"Really 404\"\n        else:\n            was_called.append('failed')\n            return 404, headers, \"Failed\"\n\n    httpretty.register_uri(httpretty.POST, url, body=request_post_callback)\n\n    # Also we want to test how would it work if cookie is available (may be)\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"),\n                                          failure_re=\"Failed\")\n\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    # first one goes with regular DownloadError -- was 404 with not matching content\n    assert_raises(DownloadError, downloader.download, url, path=d)\n    assert_equal(was_called, exp_called)\n\n\nclass FakeCredential2(UserPassword):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'user': 'testlogin', 'password': 'testpassword'}\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials\n    def enter_new(self):\n        return self._fixed_credentials\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db(cookies={'example.com': dict(some_site_id='idsomething', expires='Tue, 15 Jan 2013 21:47:38 GMT')})\ndef test_scenario_2(d):\n    fpath = opj(d, 'crap.txt')\n\n    credential = FakeCredential2(name='test', url=None)\n    credentials = credential()\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"))\n\n    def request_get_with_expired_cookie_callback(request, uri, headers):\n        assert_in('Cookie', request.headers)\n        cookie_vals = request.headers['Cookie'].split('; ')\n        for v in cookie_vals:\n            if v.startswith('expires'):\n                expiration_date = v.split('=')[1]\n                expiration_epoch_time = timegm(time.strptime(expiration_date, \"%a, %d %b %Y %H:%M:%S GMT\"))\n                assert_greater(time.time(), expiration_epoch_time)\n        return (403, headers, \"cookie was expired\")\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        assert_equal(credentials['password'], post_params['password'][0])\n        assert_equal(credentials['user'], post_params['username'][0])\n        assert_not_in('Cookie', request.headers)\n        return (200, headers, \"Got {} response from {}\".format(request.method, uri))\n\n    def request_get_callback(request, uri, headers):\n        assert_equal(request.body, b'')\n        assert_in('Cookie', request.headers)\n        assert_equal(request.headers['Cookie'], test_cookie)\n        return (200, headers, \"correct body\")\n\n    # SCENARIO 2\n    # outdated cookie provided to GET, return 403 (access denied)\n    # then like SCENARIO 1 again:\n    # POST credentials and get a new cookie\n    # which is then provided to a GET request\n    httpretty.register_uri(httpretty.GET, url,\n                           responses=[httpretty.Response(body=request_get_with_expired_cookie_callback),\n                                      httpretty.Response(body=request_get_callback),\n                                     ])\n\n    # callback to verify that correct credentials are provided\n    # and then returns the cookie to test again for 'GET'ing\n    httpretty.register_uri(httpretty.POST, url,\n                           body=request_post_callback,\n                           set_cookie=test_cookie)\n    # then in another GET is performed to verify that correct cookie was provided and\n    # that no credentials are there\n\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n\nclass FakeCredential3(Token):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'token' : 'testtoken' }\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials\n    def enter_new(self):\n        return self._fixed_credentials\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTTPBearerTokenAuthenticator(d):\n    fpath = opj(d, 'crap.txt')\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n\n\n    credential = FakeCredential3(name='test', url=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    # Perform assertions. See note above.\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken\")\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\nclass FakeLorisCredential(Token):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'token' : 'testtoken' }\n    def is_known(self):\n        return False\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTTPLorisTokenAuthenticator(d):\n    fpath = opj(d, 'crap.txt')\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n\n\n    credential = FakeCredential3(name='test', url=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    # Perform assertions. See note above.\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken\")\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\n@with_memory_keyring\n@with_testsui(responses=['yes', 'user'])\ndef test_lorisadapter(d, keyring):\n    fpath = opj(d, 'crap.txt')\n    loginurl = \"http://www.example.com/api/v0.0.2/login\"\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n    def request_post_callback(request, uri, headers):\n        return (200, headers, '{ \"token\": \"testtoken33\" }')\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n    httpretty.register_uri(httpretty.POST, loginurl,\n                           body=request_post_callback)\n\n\n\n    credential = LORIS_Token(name='test', url=loginurl, keyring=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken33\")\n    # Verify credentials correctly set to test user:pass\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n",
            "new_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Tests for http downloader\"\"\"\n\nimport time\nfrom calendar import timegm\nimport re\n\nimport os\nimport builtins\nfrom os.path import join as opj\n\nfrom datalad.downloaders.tests.utils import get_test_providers\nfrom ..base import (\n    BaseDownloader,\n    DownloadError,\n    IncompleteDownloadError,\n    NoneAuthenticator,\n)\nfrom ..credentials import (\n    LORIS_Token,\n    Token,\n    UserPassword,\n)\nfrom ..http import (\n    HTMLFormAuthenticator,\n    HTTPBaseAuthenticator,\n    HTTPDownloader,\n    HTTPBearerTokenAuthenticator,\n    process_www_authenticate,\n)\nfrom ...support.exceptions import AccessFailedError\nfrom ...support.network import get_url_straight_filename\nfrom ...utils import ensure_unicode\n\n# BTW -- mock_open is not in mock on wheezy (Debian 7.x)\ntry:\n    import httpretty\nexcept (ImportError, AttributeError):\n    # Attribute Error happens with newer httpretty and older ssl module\n    # https://github.com/datalad/datalad/pull/2623\n    class NoHTTPPretty(object):\n       __bool__ = lambda s: False\n       activate = lambda s, t: t\n    httpretty = NoHTTPPretty()\n\nfrom unittest.mock import patch\nfrom ...tests.utils import (\n    SkipTest,\n    assert_equal,\n    assert_false,\n    assert_greater,\n    assert_in,\n    assert_not_in,\n    assert_raises,\n    known_failure_githubci_win,\n    ok_file_has_content,\n    serve_path_via_http, with_tree,\n    skip_if,\n    skip_if_no_network,\n    swallow_logs,\n    swallow_outputs,\n    use_cassette,\n    with_fake_cookies_db,\n    with_memory_keyring,\n    with_tempfile,\n    with_testsui,\n    without_http_proxy,\n)\nfrom ...support.exceptions import (\n    AccessDeniedError,\n    AnonymousAccessDeniedError,\n)\nfrom ...support.status import FileStatus\nfrom ...support.network import get_url_disposition_filename\n\n\ndef test_docstring():\n    doc = HTTPDownloader.__init__.__doc__\n    assert_in(\"\\ncredential: Credential\", doc)\n\n\n# XXX doesn't quite work as it should since doesn't provide context handling\n# I guess... but at least causes the DownloadError ;)\n_builtins_open = builtins.open\n\n\ndef fake_open(write_=None, skip_regex=None):\n    class myfile(object):\n        \"\"\"file which does nothing\"\"\"\n        if write_:\n            def write(self, *args, **kwargs):\n                write_(*args, **kwargs)\n        def close(self):\n            pass\n\n    def myopen(path, *args, **kwargs):\n        if skip_regex and re.search(skip_regex, ensure_unicode(path)):\n            return _builtins_open(path, *args, **kwargs)\n        else:\n            return myfile\n    return myopen\n\n\ndef _raise_IOError(*args, **kwargs):\n    raise IOError(\"Testing here\")\n\n\ndef test_process_www_authenticate():\n    assert_equal(process_www_authenticate(\"Basic\"),\n                 [\"http_basic_auth\"])\n    assert_equal(process_www_authenticate(\"Digest\"),\n                 [\"http_digest_auth\"])\n    assert_equal(process_www_authenticate(\"Digest more\"),\n                 [\"http_digest_auth\"])\n    assert_equal(process_www_authenticate(\"Unknown\"),\n                 [])\n\n\n@known_failure_githubci_win\n@with_tree(tree=[('file.dat', 'abc')])\n@serve_path_via_http\ndef test_HTTPDownloader_basic(toppath, topurl):\n    furl = \"%sfile.dat\" % topurl\n    tfpath = opj(toppath, \"file-downloaded.dat\")\n    downloader = HTTPDownloader()  # no auth/credentials needed\n    download = downloader.download\n    download(furl, tfpath)\n    ok_file_has_content(tfpath, 'abc')\n\n    # download() creates leading directories if needed for file targets...\n    subdir_tfpath = opj(toppath, \"l1\", \"l2\", \"file-downloaded.dat\")\n    download(furl, subdir_tfpath)\n    ok_file_has_content(subdir_tfpath, 'abc')\n\n    # ... and for directory targets.\n    subdir_dirtarget = opj(toppath, \"d1\", \"d2\", \"\")\n    download(furl, subdir_dirtarget)\n    ok_file_has_content(opj(subdir_dirtarget, \"file.dat\"), \"abc\")\n\n    # see if fetch works correctly\n    assert_equal(downloader.fetch(furl), 'abc')\n\n    # By default should not overwrite the file\n    assert_raises(DownloadError, download, furl, tfpath)\n    # but be able to redownload whenever overwrite==True\n    downloaded_path = download(furl, tfpath, overwrite=True)\n    assert_equal(downloaded_path, tfpath)\n    ok_file_has_content(tfpath, 'abc')\n\n    # Fail with an informative message if we're downloading into a directory\n    # and the file name can't be determined from the URL.\n    with assert_raises(DownloadError) as cm:\n        download(topurl, toppath)\n    assert_in(\"File name could not be determined\", str(cm.exception))\n\n    # Some errors handling\n    # XXX obscure mocking since impossible to mock write alone\n    # and it still results in some warning being spit out\n    # Note: we need to avoid mocking opening of the lock file!\n    with swallow_logs(), \\\n         patch.object(builtins, 'open', fake_open(write_=_raise_IOError, skip_regex='.*\\.lck$')):\n        assert_raises(DownloadError, download, furl, tfpath, overwrite=True)\n\n    # incomplete download scenario - should have 3 tries\n    def _fail_verify_download(try_to_fail):\n        try_ = [0]\n        _orig_verify_download = BaseDownloader._verify_download\n        def _verify_download(self, *args, **kwargs):\n            try_[0] += 1\n            if try_[0] >= try_to_fail:\n                return _orig_verify_download(self, *args, **kwargs)\n            raise IncompleteDownloadError()\n        return _verify_download\n\n    with patch.object(BaseDownloader, '_verify_download', _fail_verify_download(6)), \\\n        swallow_logs():\n            # how was before the \"fix\":\n            #assert_raises(DownloadError, downloader.fetch, furl)\n            #assert_raises(DownloadError, downloader.fetch, furl)\n            # now should download just fine\n            assert_equal(downloader.fetch(furl), 'abc')\n    # but should fail if keeps failing all 5 times and then on 11th should raise DownloadError\n    with patch.object(BaseDownloader, '_verify_download', _fail_verify_download(7)), \\\n        swallow_logs():\n            assert_raises(DownloadError, downloader.fetch, furl)\n\n    # TODO: access denied scenario\n    # TODO: access denied detection\n\n\n@with_tree(tree=[('file.dat', 'abc')])\n@serve_path_via_http\n@with_memory_keyring\ndef test_access_denied(toppath, topurl, keyring):\n    furl = topurl + \"file.dat\"\n\n    def deny_access(*args, **kwargs):\n        raise AccessDeniedError(supported_types=[\"http_basic_auth\"])\n\n    def deny_anon_access(*args, **kwargs):\n        raise AnonymousAccessDeniedError(supported_types=[\"http_basic_auth\"])\n\n    downloader = HTTPDownloader()\n\n    # Test different paths that should lead to a DownloadError.\n\n    for denier in deny_access, deny_anon_access:\n        @with_testsui(responses=[\"no\"])\n        def run_refuse_provider_setup():\n            with patch.object(downloader, '_download', denier):\n                downloader.download(furl)\n        assert_raises(DownloadError, run_refuse_provider_setup)\n\n    downloader_creds = HTTPDownloader(credential=\"irrelevant\")\n\n    @with_testsui(responses=[\"no\"])\n    def run_refuse_creds_update():\n        with patch.object(downloader_creds, '_download', deny_access):\n            downloader_creds.download(furl)\n    assert_raises(DownloadError, run_refuse_creds_update)\n\n    downloader_noauth = HTTPDownloader(authenticator=NoneAuthenticator())\n\n    def run_noauth():\n        with patch.object(downloader_noauth, '_download', deny_access):\n            downloader_noauth.download(furl)\n    assert_raises(DownloadError, run_noauth)\n\n    # Complete setup for a new provider.\n\n    @with_testsui(responses=[\n        \"yes\",  # Set up provider?\n        # Enter provider details, but then don't save ...\n        \"newprovider\", re.escape(furl), \"http_auth\", \"user_password\", \"no\",\n        # No provider, try again?\n        \"yes\",\n        # Enter same provider detains but save this time.\n        \"newprovider\", re.escape(furl), \"http_auth\", \"user_password\", \"yes\",\n        # Enter credentials.\n        \"me\", \"mypass\"\n    ])\n    def run_set_up_provider():\n        with patch.object(downloader, '_download', deny_access):\n            downloader.download(furl)\n\n    # We've forced an AccessDenied error and then set up bogus credentials,\n    # leading to a 501 (not implemented) error.\n    assert_raises(AccessFailedError, run_set_up_provider)\n\n\n@with_tempfile(mkdir=True)\ndef check_download_external_url(url, failed_str, success_str, d, url_final=None):\n    fpath = opj(d, get_url_straight_filename(url))\n    providers = get_test_providers(url)  # url for check of credentials\n    provider = providers.get_provider(url)\n    downloader = provider.get_downloader(url)\n\n    # Download way\n    with swallow_outputs() as cmo:\n        downloaded_path = downloader.download(url, path=d)\n    assert_equal(fpath, downloaded_path)\n    with open(fpath) as f:\n        content = f.read()\n        if success_str is not None:\n            assert_in(success_str, content)\n        if failed_str is not None:\n            assert_false(failed_str in content)\n\n    # And if we specify size\n    for s in [1, 2]:\n        with swallow_outputs() as cmo:\n            downloaded_path_ = downloader.download(url, path=d, size=s, overwrite=True)\n        # should not be affected\n        assert_equal(downloaded_path, downloaded_path_)\n        with open(fpath) as f:\n            content_ = f.read()\n        assert_equal(len(content_), s)\n        assert_equal(content_, content[:s])\n\n    # Fetch way\n    content = downloader.fetch(url)\n    if success_str is not None:\n        assert_in(success_str, content)\n    if failed_str is not None:\n        assert_false(failed_str in content)\n\n    # And if we specify size\n    for s in [1, 2]:\n        with swallow_outputs() as cmo:\n            content_ = downloader.fetch(url, size=s)\n        assert_equal(len(content_), s)\n        assert_equal(content_, content[:s])\n\n    # Verify status\n    status = downloader.get_status(url)\n    assert(isinstance(status, FileStatus))\n    if not url.startswith('ftp://'):\n        # TODO introduce support for mtime into requests_ftp?\n        assert(status.mtime)\n    assert(status.size)\n\n    # Verify possible redirections\n    if url_final is None:\n        url_final = url\n    assert_equal(downloader.get_target_url(url), url_final)\n\n    # TODO -- more and more specific\n\n\n# TODO: @use_cassette is not playing nice with generators, causing\n# troubles when trying to cause test skip if no network. So disabling for now\n# https://github.com/datalad/datalad/issues/3158\n# @use_cassette('test_authenticate_external_portals', record_mode='once')\ndef test_authenticate_external_portals():\n    skip_if_no_network()\n    yield check_download_external_url, \\\n          \"https://portal.nersc.gov/project/crcns/download/alm-1/checksums.md5\", \\\n          \"<form action=\", \\\n          \"datafiles/meta_data_files.tar.gz\"\n    # seems to be gone\n    # yield check_download_external_url, \\\n    #       'https://db.humanconnectome.org/data/archive/projects/HCP_500/subjects/100307/experiments/100307_CREST/resources/100307_CREST/files/unprocessed/3T/Diffusion/100307_3T_DWI_dir97_LR.bval', \\\n    #       \"failed\", \\\n    #       \"2000 1005 2000 3000\"\n    yield check_download_external_url, \\\n          'https://db.humanconnectome.org/data/experiments/ConnectomeDB_E09797/resources/166768/files/filescans.csv', \\\n          \"failed\", \\\n          \"'Scan','FilePath'\"\ntest_authenticate_external_portals.tags = ['external-portal', 'network']\n\n\n@known_failure_githubci_win\n@skip_if_no_network\ndef test_download_ftp():\n    try:\n        import requests_ftp\n    except ImportError:\n        raise SkipTest(\"need requests_ftp\")  # TODO - make it not ad-hoc\n    try:\n        check_download_external_url(\n                  \"ftp://ftp.gnu.org/README\",\n                  None,\n                  \"This is ftp.gnu.org\"\n        )\n    except AccessFailedError as exc:\n        if 'status code 503' in str(exc):\n            raise SkipTest(\"ftp.gnu.org throws 503 when on travis (only?)\")\n        raise\n\n\n# TODO: redo smart way with mocking, to avoid unnecessary CPU waste\n@with_tree(tree={'file.dat': '1'})\n@serve_path_via_http\n@with_tempfile\ndef test_mtime(path, url, tempfile):\n    # let's set custom mtime\n    file_to_download = opj(path, 'file.dat')\n    os.utime(file_to_download, (time.time(), 1000))\n    assert_equal(os.stat(file_to_download).st_mtime, 1000)\n\n    file_url = \"%s/%s\" % (url, 'file.dat')\n    with swallow_outputs():\n        get_test_providers().download(file_url, path=tempfile)\n    assert_equal(os.stat(tempfile).st_mtime, 1000)\n\n\ndef test_get_status_from_headers():\n    # function doesn't do any value transformation ATM\n    headers = {\n        'Content-Length': '123',\n        # some other file record - we don't test content here yet\n        'Content-Disposition': 'attachment; filename=\"bogus.txt\"',\n        'Last-Modified': 'Sat, 07 Nov 2015 05:23:36 GMT'\n    }\n    headers['bogus1'] = '123'\n\n    assert_equal(\n            HTTPDownloader.get_status_from_headers(headers),\n            FileStatus(size=123, filename='bogus.txt', mtime=1446873816))\n\n    assert_equal(HTTPDownloader.get_status_from_headers({'content-lengtH': '123'}),\n                 FileStatus(size=123))\n\n    filename = 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'\n    headers_content_disposition = {\n        'Content-Disposition':\n            'Attachment;Filename=\"%s\"' % filename, }\n    assert_equal(\n        HTTPDownloader.get_status_from_headers(headers_content_disposition).filename,\n        filename)\n\n    # since we are providing full headers -- irrelevant\n    assert_equal(get_url_disposition_filename(\"http://irrelevant\", headers_content_disposition),\n                 filename)\n\n\n\n# TODO: test that download fails (even if authentication credentials are right) if form_url\n# is wrong!\n\n\nclass FakeCredential1(UserPassword):\n    \"\"\"Credential to test scenarios.\"\"\"\n    # to be reusable, and not leak across tests,\n    # we should get _fixed_credentials per instance\n    def __init__(self, *args, **kwargs):\n        super(FakeCredential1, self).__init__(*args, **kwargs)\n        self._fixed_credentials = [\n            {'user': 'testlogin', 'password': 'testpassword'},\n            {'user': 'testlogin2', 'password': 'testpassword2'},\n            {'user': 'testlogin2', 'password': 'testpassword3'}\n        ]\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials[0]\n    def enter_new(self):\n        # pop last used credential, so we would use \"new\" ones\n        del self._fixed_credentials[0]\n\n\nurl = \"http://example.com/crap.txt\"\ntest_cookie = 'somewebsite=testcookie'\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTMLFormAuthenticator_httpretty(d):\n    fpath = opj(d, 'crap.txt')\n\n    credential = FakeCredential1(name='test', url=None)\n    credentials = credential()\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        assert_equal(credentials['password'], post_params['password'][0])\n        assert_equal(credentials['user'], post_params['username'][0])\n        assert_not_in('Cookie', request.headers)\n        return (200, headers, \"Got {} response from {}\".format(request.method, uri))\n\n    def request_get_callback(request, uri, headers):\n        assert_equal(request.body, b'')\n        assert_in('Cookie', request.headers)\n        assert_equal(request.headers['Cookie'], test_cookie)\n        return (200, headers, \"correct body\")\n\n    # SCENARIO 1\n    # callback to verify that correct credentials are provided\n    # and then returns the cookie to test again for 'GET'ing\n    httpretty.register_uri(httpretty.POST, url,\n                           body=request_post_callback,\n                           set_cookie=test_cookie)\n    # then in GET verify that correct cookie was provided and\n    # that no credentials are there\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n    # SCENARIO 2\n    # outdated cookie provided to GET -- must return 403 (access denied)\n    # then our code should POST credentials again and get a new cookies\n    # which is then provided to GET\n\n    # SCENARIO 3\n    # outdated cookie\n    # outdated credentials\n    # it should ask for new credentials (FakeCredential1 already mocks for that)\n    # and then SCENARIO1 must work again\n\n    # SCENARIO 4\n    # cookie and credentials expired, user provided new bad credential\n\n    # Also we want to test how would it work if cookie is available (may be)\n    # TODO: check with correct and incorrect credential\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"))\n    # TODO: with success_re etc\n    # This is a \"success test\" which should be tested in various above scenarios\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n    # Unsuccesfull scenarios to test:\n    # the provided URL at the end 404s, or another failure (e.g. interrupted download)\n\n\n@with_memory_keyring\n@with_testsui(responses=['no', 'yes', 'testlogin', 'testpassword'])\ndef test_auth_but_no_cred(keyring):\n    authenticator = HTMLFormAuthenticator(\"\")\n    # Replying 'no' to the set credentials prompt should raise ValueError\n    assert_raises(ValueError, HTTPDownloader, credential=None, authenticator=authenticator)\n    # Reply 'yes' and set test user:pass at the next set credentials prompt\n    downloader = HTTPDownloader(credential=None, authenticator=authenticator)\n    # Verify credentials correctly set to test user:pass\n    assert_equal(downloader.credential.get('user'), 'testlogin')\n    assert_equal(downloader.credential.get('password'), 'testpassword')\n\n\n@with_testsui(responses=['yes'])  # will request to reentry it\ndef test_authfail404_interactive():\n    # we will firsts get 'failed' but then real 404 when trying new password\n    check_httpretty_authfail404(['failed', '404'])\n\n\n@with_testsui(interactive=False)  # no interactions -- blow!\ndef test_authfail404_noninteractive():\n    # we do not get to the 2nd attempt so just get 'failed'\n    # and exception thrown inside is not emerging all the way here but\n    # caught in the check_\n    check_httpretty_authfail404(['failed'])\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_fake_cookies_db\n@with_tempfile(mkdir=True)\ndef check_httpretty_authfail404(exp_called, d):\n    # mimic behavior of nersc which 404s but provides feedback whenever\n    # credentials are incorrect.  In our case we should fail properly\n    credential = FakeCredential1(name='test', url=None)\n\n    was_called = []\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        if post_params['password'][0] == 'testpassword2':\n            was_called.append('404')\n            return 404, headers, \"Really 404\"\n        else:\n            was_called.append('failed')\n            return 404, headers, \"Failed\"\n\n    httpretty.register_uri(httpretty.POST, url, body=request_post_callback)\n\n    # Also we want to test how would it work if cookie is available (may be)\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"),\n                                          failure_re=\"Failed\")\n\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    # first one goes with regular DownloadError -- was 404 with not matching content\n    assert_raises(DownloadError, downloader.download, url, path=d)\n    assert_equal(was_called, exp_called)\n\n\ndef test_auth_bytes_content():\n    # Our regexes are strings, but we can get content in bytes:\n    # I am not sure yet either we shouldn't just skip then testing for regex,\n    # but we definetely should not crash.\n    authenticator = HTTPBaseAuthenticator(failure_re=\"Failed\")\n    authenticator.check_for_auth_failure(b\"bytes\")\n    # but ATM we do test bytes content, let's ENSURE that!\n    with assert_raises(AccessDeniedError):\n        authenticator.check_for_auth_failure(b\"Failed\")\n\n\nclass FakeCredential2(UserPassword):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'user': 'testlogin', 'password': 'testpassword'}\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials\n    def enter_new(self):\n        return self._fixed_credentials\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db(cookies={'example.com': dict(some_site_id='idsomething', expires='Tue, 15 Jan 2013 21:47:38 GMT')})\ndef test_scenario_2(d):\n    fpath = opj(d, 'crap.txt')\n\n    credential = FakeCredential2(name='test', url=None)\n    credentials = credential()\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"))\n\n    def request_get_with_expired_cookie_callback(request, uri, headers):\n        assert_in('Cookie', request.headers)\n        cookie_vals = request.headers['Cookie'].split('; ')\n        for v in cookie_vals:\n            if v.startswith('expires'):\n                expiration_date = v.split('=')[1]\n                expiration_epoch_time = timegm(time.strptime(expiration_date, \"%a, %d %b %Y %H:%M:%S GMT\"))\n                assert_greater(time.time(), expiration_epoch_time)\n        return (403, headers, \"cookie was expired\")\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        assert_equal(credentials['password'], post_params['password'][0])\n        assert_equal(credentials['user'], post_params['username'][0])\n        assert_not_in('Cookie', request.headers)\n        return (200, headers, \"Got {} response from {}\".format(request.method, uri))\n\n    def request_get_callback(request, uri, headers):\n        assert_equal(request.body, b'')\n        assert_in('Cookie', request.headers)\n        assert_equal(request.headers['Cookie'], test_cookie)\n        return (200, headers, \"correct body\")\n\n    # SCENARIO 2\n    # outdated cookie provided to GET, return 403 (access denied)\n    # then like SCENARIO 1 again:\n    # POST credentials and get a new cookie\n    # which is then provided to a GET request\n    httpretty.register_uri(httpretty.GET, url,\n                           responses=[httpretty.Response(body=request_get_with_expired_cookie_callback),\n                                      httpretty.Response(body=request_get_callback),\n                                     ])\n\n    # callback to verify that correct credentials are provided\n    # and then returns the cookie to test again for 'GET'ing\n    httpretty.register_uri(httpretty.POST, url,\n                           body=request_post_callback,\n                           set_cookie=test_cookie)\n    # then in another GET is performed to verify that correct cookie was provided and\n    # that no credentials are there\n\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n\nclass FakeCredential3(Token):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'token' : 'testtoken' }\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials\n    def enter_new(self):\n        return self._fixed_credentials\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTTPBearerTokenAuthenticator(d):\n    fpath = opj(d, 'crap.txt')\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n\n\n    credential = FakeCredential3(name='test', url=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    # Perform assertions. See note above.\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken\")\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\nclass FakeLorisCredential(Token):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'token' : 'testtoken' }\n    def is_known(self):\n        return False\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTTPLorisTokenAuthenticator(d):\n    fpath = opj(d, 'crap.txt')\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n\n\n    credential = FakeCredential3(name='test', url=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    # Perform assertions. See note above.\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken\")\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\n@with_memory_keyring\n@with_testsui(responses=['yes', 'user'])\ndef test_lorisadapter(d, keyring):\n    fpath = opj(d, 'crap.txt')\n    loginurl = \"http://www.example.com/api/v0.0.2/login\"\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n    def request_post_callback(request, uri, headers):\n        return (200, headers, '{ \"token\": \"testtoken33\" }')\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n    httpretty.register_uri(httpretty.POST, loginurl,\n                           body=request_post_callback)\n\n\n\n    credential = LORIS_Token(name='test', url=loginurl, keyring=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken33\")\n    # Verify credentials correctly set to test user:pass\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n",
            "header": {
                "file": {
                    "path": "datalad/downloaders/tests/test_http.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "dbbd0864d",
                "new_commit_hash": "ce1d744c1",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/downloaders/tests/test_http.py"
            },
            "plus_file": {
                "path": "b/datalad/downloaders/tests/test_http.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 30,
                            "length": 6
                        },
                        "new_range": {
                            "start": 30,
                            "length": 7
                        },
                        "section": "from ..credentials import ("
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": ")",
                                "type": "context"
                            },
                            {
                                "content": "from ..http import (",
                                "type": "context"
                            },
                            {
                                "content": "    HTMLFormAuthenticator,",
                                "type": "context"
                            },
                            {
                                "content": "    HTTPBaseAuthenticator,",
                                "type": "added"
                            },
                            {
                                "content": "    HTTPDownloader,",
                                "type": "context"
                            },
                            {
                                "content": "    HTTPBearerTokenAuthenticator,",
                                "type": "context"
                            },
                            {
                                "content": "    process_www_authenticate,",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "datalad/downloaders/tests/test_http.py",
                            "type": "statement",
                            "name": "",
                            "content": "from ..http import (\n    HTMLFormAuthenticator,\n    HTTPBaseAuthenticator,\n    HTTPDownloader,\n    HTTPBearerTokenAuthenticator,\n    process_www_authenticate,\n)",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 31,
                            "end_lineno": 37,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 558,
                            "length": 6
                        },
                        "new_range": {
                            "start": 559,
                            "length": 17
                        },
                        "section": "def check_httpretty_authfail404(exp_called, d):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    assert_equal(was_called, exp_called)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "def test_auth_bytes_content():",
                                "type": "added"
                            },
                            {
                                "content": "    # Our regexes are strings, but we can get content in bytes:",
                                "type": "added"
                            },
                            {
                                "content": "    # I am not sure yet either we shouldn't just skip then testing for regex,",
                                "type": "added"
                            },
                            {
                                "content": "    # but we definetely should not crash.",
                                "type": "added"
                            },
                            {
                                "content": "    authenticator = HTTPBaseAuthenticator(failure_re=\"Failed\")",
                                "type": "added"
                            },
                            {
                                "content": "    authenticator.check_for_auth_failure(b\"bytes\")",
                                "type": "added"
                            },
                            {
                                "content": "    # but ATM we do test bytes content, let's ENSURE that!",
                                "type": "added"
                            },
                            {
                                "content": "    with assert_raises(AccessDeniedError):",
                                "type": "added"
                            },
                            {
                                "content": "        authenticator.check_for_auth_failure(b\"Failed\")",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "class FakeCredential2(UserPassword):",
                                "type": "context"
                            },
                            {
                                "content": "    \"\"\"Credential to test scenarios.\"\"\"",
                                "type": "context"
                            },
                            {
                                "content": "    _fixed_credentials = {'user': 'testlogin', 'password': 'testpassword'}",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "datalad/downloaders/tests/test_http.py",
                            "type": "function",
                            "name": "test_auth_bytes_content",
                            "content": "def test_auth_bytes_content():\n    # Our regexes are strings, but we can get content in bytes:\n    # I am not sure yet either we shouldn't just skip then testing for regex,\n    # but we definetely should not crash.\n    authenticator = HTTPBaseAuthenticator(failure_re=\"Failed\")\n    authenticator.check_for_auth_failure(b\"bytes\")\n    # but ATM we do test bytes content, let's ENSURE that!\n    with assert_raises(AccessDeniedError):\n        authenticator.check_for_auth_failure(b\"Failed\")",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 562,
                            "end_lineno": 570,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "0174e66d85d0fa42d25f1231039a7889ef964237^",
    "new_commit_hash": "0174e66d85d0fa42d25f1231039a7889ef964237",
    "commit_message": "BF: if content comes as bytes, ensure that our regex is bytes as well\n\nSee 3b55eb4e9d23fc22d175b0cfa9f1a8320b2258be for more details.",
    "commit_date": "2020-05-15T16:44:54-04:00",
    "metadata": {}
}
{
    "repo_name": "datalad",
    "new_commit_hash": "0174e66d85d0fa42d25f1231039a7889ef964237",
    "test_file_codes": [
        "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Tests for http downloader\"\"\"\n\nimport time\nfrom calendar import timegm\nimport re\n\nimport os\nimport builtins\nfrom os.path import join as opj\n\nfrom datalad.downloaders.tests.utils import get_test_providers\nfrom datalad.downloaders.base import (\n    BaseDownloader,\n    DownloadError,\n    IncompleteDownloadError,\n    NoneAuthenticator,\n)\nfrom datalad.downloaders.credentials import (\n    LORIS_Token,\n    Token,\n    UserPassword,\n)\nfrom datalad.downloaders.http import (\n    HTMLFormAuthenticator,\n    HTTPBaseAuthenticator,\n    HTTPDownloader,\n    HTTPBearerTokenAuthenticator,\n    process_www_authenticate,\n)\nfrom datalad.support.exceptions import AccessFailedError\nfrom datalad.support.network import get_url_straight_filename\nfrom datalad.utils import ensure_unicode\n\n# BTW -- mock_open is not in mock on wheezy (Debian 7.x)\ntry:\n    import httpretty\nexcept (ImportError, AttributeError):\n    # Attribute Error happens with newer httpretty and older ssl module\n    # https://github.com/datalad/datalad/pull/2623\n    class NoHTTPPretty(object):\n       __bool__ = lambda s: False\n       activate = lambda s, t: t\n    httpretty = NoHTTPPretty()\n\nfrom unittest.mock import patch\nfrom datalad.tests.utils import (\n    SkipTest,\n    assert_equal,\n    assert_false,\n    assert_greater,\n    assert_in,\n    assert_not_in,\n    assert_raises,\n    known_failure_githubci_win,\n    ok_file_has_content,\n    serve_path_via_http, with_tree,\n    skip_if,\n    skip_if_no_network,\n    swallow_logs,\n    swallow_outputs,\n    use_cassette,\n    with_fake_cookies_db,\n    with_memory_keyring,\n    with_tempfile,\n    with_testsui,\n    without_http_proxy,\n)\nfrom datalad.support.exceptions import (\n    AccessDeniedError,\n    AnonymousAccessDeniedError,\n)\nfrom datalad.support.status import FileStatus\nfrom datalad.support.network import get_url_disposition_filename\n\n\ndef test_docstring():\n    doc = HTTPDownloader.__init__.__doc__\n    assert_in(\"\\ncredential: Credential\", doc)\n\n\n# XXX doesn't quite work as it should since doesn't provide context handling\n# I guess... but at least causes the DownloadError ;)\n_builtins_open = builtins.open\n\n\ndef fake_open(write_=None, skip_regex=None):\n    class myfile(object):\n        \"\"\"file which does nothing\"\"\"\n        if write_:\n            def write(self, *args, **kwargs):\n                write_(*args, **kwargs)\n        def close(self):\n            pass\n\n    def myopen(path, *args, **kwargs):\n        if skip_regex and re.search(skip_regex, ensure_unicode(path)):\n            return _builtins_open(path, *args, **kwargs)\n        else:\n            return myfile\n    return myopen\n\n\ndef _raise_IOError(*args, **kwargs):\n    raise IOError(\"Testing here\")\n\n\ndef test_process_www_authenticate():\n    assert_equal(process_www_authenticate(\"Basic\"),\n                 [\"http_basic_auth\"])\n    assert_equal(process_www_authenticate(\"Digest\"),\n                 [\"http_digest_auth\"])\n    assert_equal(process_www_authenticate(\"Digest more\"),\n                 [\"http_digest_auth\"])\n    assert_equal(process_www_authenticate(\"Unknown\"),\n                 [])\n\n\n@known_failure_githubci_win\n@with_tree(tree=[('file.dat', 'abc')])\n@serve_path_via_http\ndef test_HTTPDownloader_basic(toppath, topurl):\n    furl = \"%sfile.dat\" % topurl\n    tfpath = opj(toppath, \"file-downloaded.dat\")\n    downloader = HTTPDownloader()  # no auth/credentials needed\n    download = downloader.download\n    download(furl, tfpath)\n    ok_file_has_content(tfpath, 'abc')\n\n    # download() creates leading directories if needed for file targets...\n    subdir_tfpath = opj(toppath, \"l1\", \"l2\", \"file-downloaded.dat\")\n    download(furl, subdir_tfpath)\n    ok_file_has_content(subdir_tfpath, 'abc')\n\n    # ... and for directory targets.\n    subdir_dirtarget = opj(toppath, \"d1\", \"d2\", \"\")\n    download(furl, subdir_dirtarget)\n    ok_file_has_content(opj(subdir_dirtarget, \"file.dat\"), \"abc\")\n\n    # see if fetch works correctly\n    assert_equal(downloader.fetch(furl), 'abc')\n\n    # By default should not overwrite the file\n    assert_raises(DownloadError, download, furl, tfpath)\n    # but be able to redownload whenever overwrite==True\n    downloaded_path = download(furl, tfpath, overwrite=True)\n    assert_equal(downloaded_path, tfpath)\n    ok_file_has_content(tfpath, 'abc')\n\n    # Fail with an informative message if we're downloading into a directory\n    # and the file name can't be determined from the URL.\n    with assert_raises(DownloadError) as cm:\n        download(topurl, toppath)\n    assert_in(\"File name could not be determined\", str(cm.exception))\n\n    # Some errors handling\n    # XXX obscure mocking since impossible to mock write alone\n    # and it still results in some warning being spit out\n    # Note: we need to avoid mocking opening of the lock file!\n    with swallow_logs(), \\\n         patch.object(builtins, 'open', fake_open(write_=_raise_IOError, skip_regex='.*\\.lck$')):\n        assert_raises(DownloadError, download, furl, tfpath, overwrite=True)\n\n    # incomplete download scenario - should have 3 tries\n    def _fail_verify_download(try_to_fail):\n        try_ = [0]\n        _orig_verify_download = BaseDownloader._verify_download\n        def _verify_download(self, *args, **kwargs):\n            try_[0] += 1\n            if try_[0] >= try_to_fail:\n                return _orig_verify_download(self, *args, **kwargs)\n            raise IncompleteDownloadError()\n        return _verify_download\n\n    with patch.object(BaseDownloader, '_verify_download', _fail_verify_download(6)), \\\n        swallow_logs():\n            # how was before the \"fix\":\n            #assert_raises(DownloadError, downloader.fetch, furl)\n            #assert_raises(DownloadError, downloader.fetch, furl)\n            # now should download just fine\n            assert_equal(downloader.fetch(furl), 'abc')\n    # but should fail if keeps failing all 5 times and then on 11th should raise DownloadError\n    with patch.object(BaseDownloader, '_verify_download', _fail_verify_download(7)), \\\n        swallow_logs():\n            assert_raises(DownloadError, downloader.fetch, furl)\n\n    # TODO: access denied scenario\n    # TODO: access denied detection\n\n\n@with_tree(tree=[('file.dat', 'abc')])\n@serve_path_via_http\n@with_memory_keyring\ndef test_access_denied(toppath, topurl, keyring):\n    furl = topurl + \"file.dat\"\n\n    def deny_access(*args, **kwargs):\n        raise AccessDeniedError(supported_types=[\"http_basic_auth\"])\n\n    def deny_anon_access(*args, **kwargs):\n        raise AnonymousAccessDeniedError(supported_types=[\"http_basic_auth\"])\n\n    downloader = HTTPDownloader()\n\n    # Test different paths that should lead to a DownloadError.\n\n    for denier in deny_access, deny_anon_access:\n        @with_testsui(responses=[\"no\"])\n        def run_refuse_provider_setup():\n            with patch.object(downloader, '_download', denier):\n                downloader.download(furl)\n        assert_raises(DownloadError, run_refuse_provider_setup)\n\n    downloader_creds = HTTPDownloader(credential=\"irrelevant\")\n\n    @with_testsui(responses=[\"no\"])\n    def run_refuse_creds_update():\n        with patch.object(downloader_creds, '_download', deny_access):\n            downloader_creds.download(furl)\n    assert_raises(DownloadError, run_refuse_creds_update)\n\n    downloader_noauth = HTTPDownloader(authenticator=NoneAuthenticator())\n\n    def run_noauth():\n        with patch.object(downloader_noauth, '_download', deny_access):\n            downloader_noauth.download(furl)\n    assert_raises(DownloadError, run_noauth)\n\n    # Complete setup for a new provider.\n\n    @with_testsui(responses=[\n        \"yes\",  # Set up provider?\n        # Enter provider details, but then don't save ...\n        \"newprovider\", re.escape(furl), \"http_auth\", \"user_password\", \"no\",\n        # No provider, try again?\n        \"yes\",\n        # Enter same provider detains but save this time.\n        \"newprovider\", re.escape(furl), \"http_auth\", \"user_password\", \"yes\",\n        # Enter credentials.\n        \"me\", \"mypass\"\n    ])\n    def run_set_up_provider():\n        with patch.object(downloader, '_download', deny_access):\n            downloader.download(furl)\n\n    # We've forced an AccessDenied error and then set up bogus credentials,\n    # leading to a 501 (not implemented) error.\n    assert_raises(AccessFailedError, run_set_up_provider)\n\n\n@with_tempfile(mkdir=True)\ndef check_download_external_url(url, failed_str, success_str, d, url_final=None):\n    fpath = opj(d, get_url_straight_filename(url))\n    providers = get_test_providers(url)  # url for check of credentials\n    provider = providers.get_provider(url)\n    downloader = provider.get_downloader(url)\n\n    # Download way\n    with swallow_outputs() as cmo:\n        downloaded_path = downloader.download(url, path=d)\n    assert_equal(fpath, downloaded_path)\n    with open(fpath) as f:\n        content = f.read()\n        if success_str is not None:\n            assert_in(success_str, content)\n        if failed_str is not None:\n            assert_false(failed_str in content)\n\n    # And if we specify size\n    for s in [1, 2]:\n        with swallow_outputs() as cmo:\n            downloaded_path_ = downloader.download(url, path=d, size=s, overwrite=True)\n        # should not be affected\n        assert_equal(downloaded_path, downloaded_path_)\n        with open(fpath) as f:\n            content_ = f.read()\n        assert_equal(len(content_), s)\n        assert_equal(content_, content[:s])\n\n    # Fetch way\n    content = downloader.fetch(url)\n    if success_str is not None:\n        assert_in(success_str, content)\n    if failed_str is not None:\n        assert_false(failed_str in content)\n\n    # And if we specify size\n    for s in [1, 2]:\n        with swallow_outputs() as cmo:\n            content_ = downloader.fetch(url, size=s)\n        assert_equal(len(content_), s)\n        assert_equal(content_, content[:s])\n\n    # Verify status\n    status = downloader.get_status(url)\n    assert(isinstance(status, FileStatus))\n    if not url.startswith('ftp://'):\n        # TODO introduce support for mtime into requests_ftp?\n        assert(status.mtime)\n    assert(status.size)\n\n    # Verify possible redirections\n    if url_final is None:\n        url_final = url\n    assert_equal(downloader.get_target_url(url), url_final)\n\n    # TODO -- more and more specific\n\n\n# TODO: @use_cassette is not playing nice with generators, causing\n# troubles when trying to cause test skip if no network. So disabling for now\n# https://github.com/datalad/datalad/issues/3158\n# @use_cassette('test_authenticate_external_portals', record_mode='once')\ndef test_authenticate_external_portals():\n    skip_if_no_network()\n    yield check_download_external_url, \\\n          \"https://portal.nersc.gov/project/crcns/download/alm-1/checksums.md5\", \\\n          \"<form action=\", \\\n          \"datafiles/meta_data_files.tar.gz\"\n    # seems to be gone\n    # yield check_download_external_url, \\\n    #       'https://db.humanconnectome.org/data/archive/projects/HCP_500/subjects/100307/experiments/100307_CREST/resources/100307_CREST/files/unprocessed/3T/Diffusion/100307_3T_DWI_dir97_LR.bval', \\\n    #       \"failed\", \\\n    #       \"2000 1005 2000 3000\"\n    yield check_download_external_url, \\\n          'https://db.humanconnectome.org/data/experiments/ConnectomeDB_E09797/resources/166768/files/filescans.csv', \\\n          \"failed\", \\\n          \"'Scan','FilePath'\"\ntest_authenticate_external_portals.tags = ['external-portal', 'network']\n\n\n@known_failure_githubci_win\n@skip_if_no_network\ndef test_download_ftp():\n    try:\n        import requests_ftp\n    except ImportError:\n        raise SkipTest(\"need requests_ftp\")  # TODO - make it not ad-hoc\n    try:\n        check_download_external_url(\n                  \"ftp://ftp.gnu.org/README\",\n                  None,\n                  \"This is ftp.gnu.org\"\n        )\n    except AccessFailedError as exc:\n        if 'status code 503' in str(exc):\n            raise SkipTest(\"ftp.gnu.org throws 503 when on travis (only?)\")\n        raise\n\n\n# TODO: redo smart way with mocking, to avoid unnecessary CPU waste\n@with_tree(tree={'file.dat': '1'})\n@serve_path_via_http\n@with_tempfile\ndef test_mtime(path, url, tempfile):\n    # let's set custom mtime\n    file_to_download = opj(path, 'file.dat')\n    os.utime(file_to_download, (time.time(), 1000))\n    assert_equal(os.stat(file_to_download).st_mtime, 1000)\n\n    file_url = \"%s/%s\" % (url, 'file.dat')\n    with swallow_outputs():\n        get_test_providers().download(file_url, path=tempfile)\n    assert_equal(os.stat(tempfile).st_mtime, 1000)\n\n\ndef test_get_status_from_headers():\n    # function doesn't do any value transformation ATM\n    headers = {\n        'Content-Length': '123',\n        # some other file record - we don't test content here yet\n        'Content-Disposition': 'attachment; filename=\"bogus.txt\"',\n        'Last-Modified': 'Sat, 07 Nov 2015 05:23:36 GMT'\n    }\n    headers['bogus1'] = '123'\n\n    assert_equal(\n            HTTPDownloader.get_status_from_headers(headers),\n            FileStatus(size=123, filename='bogus.txt', mtime=1446873816))\n\n    assert_equal(HTTPDownloader.get_status_from_headers({'content-lengtH': '123'}),\n                 FileStatus(size=123))\n\n    filename = 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'\n    headers_content_disposition = {\n        'Content-Disposition':\n            'Attachment;Filename=\"%s\"' % filename, }\n    assert_equal(\n        HTTPDownloader.get_status_from_headers(headers_content_disposition).filename,\n        filename)\n\n    # since we are providing full headers -- irrelevant\n    assert_equal(get_url_disposition_filename(\"http://irrelevant\", headers_content_disposition),\n                 filename)\n\n\n\n# TODO: test that download fails (even if authentication credentials are right) if form_url\n# is wrong!\n\n\nclass FakeCredential1(UserPassword):\n    \"\"\"Credential to test scenarios.\"\"\"\n    # to be reusable, and not leak across tests,\n    # we should get _fixed_credentials per instance\n    def __init__(self, *args, **kwargs):\n        super(FakeCredential1, self).__init__(*args, **kwargs)\n        self._fixed_credentials = [\n            {'user': 'testlogin', 'password': 'testpassword'},\n            {'user': 'testlogin2', 'password': 'testpassword2'},\n            {'user': 'testlogin2', 'password': 'testpassword3'}\n        ]\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials[0]\n    def enter_new(self):\n        # pop last used credential, so we would use \"new\" ones\n        del self._fixed_credentials[0]\n\n\nurl = \"http://example.com/crap.txt\"\ntest_cookie = 'somewebsite=testcookie'\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTMLFormAuthenticator_httpretty(d):\n    fpath = opj(d, 'crap.txt')\n\n    credential = FakeCredential1(name='test', url=None)\n    credentials = credential()\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        assert_equal(credentials['password'], post_params['password'][0])\n        assert_equal(credentials['user'], post_params['username'][0])\n        assert_not_in('Cookie', request.headers)\n        return (200, headers, \"Got {} response from {}\".format(request.method, uri))\n\n    def request_get_callback(request, uri, headers):\n        assert_equal(request.body, b'')\n        assert_in('Cookie', request.headers)\n        assert_equal(request.headers['Cookie'], test_cookie)\n        return (200, headers, \"correct body\")\n\n    # SCENARIO 1\n    # callback to verify that correct credentials are provided\n    # and then returns the cookie to test again for 'GET'ing\n    httpretty.register_uri(httpretty.POST, url,\n                           body=request_post_callback,\n                           set_cookie=test_cookie)\n    # then in GET verify that correct cookie was provided and\n    # that no credentials are there\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n    # SCENARIO 2\n    # outdated cookie provided to GET -- must return 403 (access denied)\n    # then our code should POST credentials again and get a new cookies\n    # which is then provided to GET\n\n    # SCENARIO 3\n    # outdated cookie\n    # outdated credentials\n    # it should ask for new credentials (FakeCredential1 already mocks for that)\n    # and then SCENARIO1 must work again\n\n    # SCENARIO 4\n    # cookie and credentials expired, user provided new bad credential\n\n    # Also we want to test how would it work if cookie is available (may be)\n    # TODO: check with correct and incorrect credential\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"))\n    # TODO: with success_re etc\n    # This is a \"success test\" which should be tested in various above scenarios\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n    # Unsuccesfull scenarios to test:\n    # the provided URL at the end 404s, or another failure (e.g. interrupted download)\n\n\n@with_memory_keyring\n@with_testsui(responses=['no', 'yes', 'testlogin', 'testpassword'])\ndef test_auth_but_no_cred(keyring):\n    authenticator = HTMLFormAuthenticator(\"\")\n    # Replying 'no' to the set credentials prompt should raise ValueError\n    assert_raises(ValueError, HTTPDownloader, credential=None, authenticator=authenticator)\n    # Reply 'yes' and set test user:pass at the next set credentials prompt\n    downloader = HTTPDownloader(credential=None, authenticator=authenticator)\n    # Verify credentials correctly set to test user:pass\n    assert_equal(downloader.credential.get('user'), 'testlogin')\n    assert_equal(downloader.credential.get('password'), 'testpassword')\n\n\n@with_testsui(responses=['yes'])  # will request to reentry it\ndef test_authfail404_interactive():\n    # we will firsts get 'failed' but then real 404 when trying new password\n    check_httpretty_authfail404(['failed', '404'])\n\n\n@with_testsui(interactive=False)  # no interactions -- blow!\ndef test_authfail404_noninteractive():\n    # we do not get to the 2nd attempt so just get 'failed'\n    # and exception thrown inside is not emerging all the way here but\n    # caught in the check_\n    check_httpretty_authfail404(['failed'])\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_fake_cookies_db\n@with_tempfile(mkdir=True)\ndef check_httpretty_authfail404(exp_called, d):\n    # mimic behavior of nersc which 404s but provides feedback whenever\n    # credentials are incorrect.  In our case we should fail properly\n    credential = FakeCredential1(name='test', url=None)\n\n    was_called = []\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        if post_params['password'][0] == 'testpassword2':\n            was_called.append('404')\n            return 404, headers, \"Really 404\"\n        else:\n            was_called.append('failed')\n            return 404, headers, \"Failed\"\n\n    httpretty.register_uri(httpretty.POST, url, body=request_post_callback)\n\n    # Also we want to test how would it work if cookie is available (may be)\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"),\n                                          failure_re=\"Failed\")\n\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    # first one goes with regular DownloadError -- was 404 with not matching content\n    assert_raises(DownloadError, downloader.download, url, path=d)\n    assert_equal(was_called, exp_called)\n\n\ndef test_auth_bytes_content():\n    # Our regexes are strings, but we can get content in bytes:\n    # I am not sure yet either we shouldn't just skip then testing for regex,\n    # but we definetely should not crash.\n    authenticator = HTTPBaseAuthenticator(failure_re=\"Failed\")\n    authenticator.check_for_auth_failure(b\"bytes\")\n    # but ATM we do test bytes content, let's ENSURE that!\n    with assert_raises(AccessDeniedError):\n        authenticator.check_for_auth_failure(b\"Failed\")\n\n\nclass FakeCredential2(UserPassword):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'user': 'testlogin', 'password': 'testpassword'}\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials\n    def enter_new(self):\n        return self._fixed_credentials\n\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db(cookies={'example.com': dict(some_site_id='idsomething', expires='Tue, 15 Jan 2013 21:47:38 GMT')})\ndef test_scenario_2(d):\n    fpath = opj(d, 'crap.txt')\n\n    credential = FakeCredential2(name='test', url=None)\n    credentials = credential()\n    authenticator = HTMLFormAuthenticator(dict(username=\"{user}\",\n                                               password=\"{password}\",\n                                               submit=\"CustomLogin\"))\n\n    def request_get_with_expired_cookie_callback(request, uri, headers):\n        assert_in('Cookie', request.headers)\n        cookie_vals = request.headers['Cookie'].split('; ')\n        for v in cookie_vals:\n            if v.startswith('expires'):\n                expiration_date = v.split('=')[1]\n                expiration_epoch_time = timegm(time.strptime(expiration_date, \"%a, %d %b %Y %H:%M:%S GMT\"))\n                assert_greater(time.time(), expiration_epoch_time)\n        return (403, headers, \"cookie was expired\")\n\n    def request_post_callback(request, uri, headers):\n        post_params = request.parsed_body\n        assert_equal(credentials['password'], post_params['password'][0])\n        assert_equal(credentials['user'], post_params['username'][0])\n        assert_not_in('Cookie', request.headers)\n        return (200, headers, \"Got {} response from {}\".format(request.method, uri))\n\n    def request_get_callback(request, uri, headers):\n        assert_equal(request.body, b'')\n        assert_in('Cookie', request.headers)\n        assert_equal(request.headers['Cookie'], test_cookie)\n        return (200, headers, \"correct body\")\n\n    # SCENARIO 2\n    # outdated cookie provided to GET, return 403 (access denied)\n    # then like SCENARIO 1 again:\n    # POST credentials and get a new cookie\n    # which is then provided to a GET request\n    httpretty.register_uri(httpretty.GET, url,\n                           responses=[httpretty.Response(body=request_get_with_expired_cookie_callback),\n                                      httpretty.Response(body=request_get_callback),\n                                     ])\n\n    # callback to verify that correct credentials are provided\n    # and then returns the cookie to test again for 'GET'ing\n    httpretty.register_uri(httpretty.POST, url,\n                           body=request_post_callback,\n                           set_cookie=test_cookie)\n    # then in another GET is performed to verify that correct cookie was provided and\n    # that no credentials are there\n\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n\nclass FakeCredential3(Token):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'token' : 'testtoken' }\n    def is_known(self):\n        return True\n    def __call__(self):\n        return self._fixed_credentials\n    def enter_new(self):\n        return self._fixed_credentials\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTTPBearerTokenAuthenticator(d):\n    fpath = opj(d, 'crap.txt')\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n\n\n    credential = FakeCredential3(name='test', url=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    # Perform assertions. See note above.\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken\")\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\nclass FakeLorisCredential(Token):\n    \"\"\"Credential to test scenarios.\"\"\"\n    _fixed_credentials = {'token' : 'testtoken' }\n    def is_known(self):\n        return False\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\ndef test_HTTPLorisTokenAuthenticator(d):\n    fpath = opj(d, 'crap.txt')\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n\n\n\n    credential = FakeCredential3(name='test', url=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    # Perform assertions. See note above.\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken\")\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n\n@skip_if(not httpretty, \"no httpretty\")\n@without_http_proxy\n@httpretty.activate\n@with_tempfile(mkdir=True)\n@with_fake_cookies_db\n@with_memory_keyring\n@with_testsui(responses=['yes', 'user'])\ndef test_lorisadapter(d, keyring):\n    fpath = opj(d, 'crap.txt')\n    loginurl = \"http://www.example.com/api/v0.0.2/login\"\n\n    def request_get_callback(request, uri, headers):\n        # We can't assert inside the callback, or running the\n        # test give \"Connection aborted\" errors instead of telling\n        # us that the assertion failed. So instead, we make\n        # the request object available outside of the callback\n        # and do the assertions in the main test, not the callback\n        request_get_callback.req = request\n        return (200, headers, \"correct body\")\n    def request_post_callback(request, uri, headers):\n        return (200, headers, '{ \"token\": \"testtoken33\" }')\n\n    httpretty.register_uri(httpretty.GET, url,\n                           body=request_get_callback)\n    httpretty.register_uri(httpretty.POST, loginurl,\n                           body=request_post_callback)\n\n\n\n    credential = LORIS_Token(name='test', url=loginurl, keyring=None)\n    authenticator = HTTPBearerTokenAuthenticator()\n    downloader = HTTPDownloader(credential=credential, authenticator=authenticator)\n    downloader.download(url, path=d)\n\n    r = request_get_callback.req\n    assert_equal(r.body, b'')\n    assert_in('Authorization', r.headers)\n    assert_equal(r.headers['Authorization'], \"Bearer testtoken33\")\n    # Verify credentials correctly set to test user:pass\n\n    with open(fpath) as f:\n        content = f.read()\n        assert_equal(content, \"correct body\")\n",
        "import pytest\nimport tempfile\nimport os\n\n\n@pytest.fixture\ndef path():\n    with tempfile.TemporaryDirectory() as tmpdir:\n        yield tmpdir\n\n\n@pytest.fixture(autouse=True)\ndef setup_git_config():\n    with tempfile.TemporaryDirectory() as home:\n        old_home = os.environ.get(\"HOME\")\n        os.environ[\"HOME\"] = home\n        os.system('git config --global user.name \"DataLad Tester\"')\n        os.system('git config --global user.email \"test@example.com\"')\n        yield\n        if old_home:\n            os.environ[\"HOME\"] = old_home\n"
    ],
    "test_file_names": [
        "test_1.py",
        "conftest.py"
    ],
    "setup_res_code": 0,
    "setup_res_stdout": "Starting Datalad installation attempts...\nVerifying installation...\nDatalad version: 0.12.6.dev29\n✅ Installation successful!\nSuccessfully installed Datalad using Python 3.9\n",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 25 packages in 73ms\nInstalled 25 packages in 5.13s\n + coverage==7.6.10\n + docopt==0.6.2\n + exceptiongroup==1.2.2\n + grabbit==0.2.6\n + importlib-resources==6.4.5\n + iniconfig==2.0.0\n + nibabel==5.3.2\n + num2words==0.5.14\n + numpy==2.0.2\n + packaging==24.2\n + pandas==2.2.3\n + patsy==1.0.1\n + pluggy==1.5.0\n + pybids==0.6.5\n + pytest==8.3.4\n + pytest-cov==6.0.0\n + python-dateutil==2.9.0.post0\n + pytz==2024.2\n + scipy==1.13.1\n + setuptools==75.6.0\n + six==1.17.0\n + tomli==2.2.1\n + typing-extensions==4.12.2\n + tzdata==2024.2\n + zipp==3.21.0\nResolved 65 packages in 4.07s\nPrepared 1 package in 3.97s\nInstalled 59 packages in 831ms\n + appdirs==1.4.4\n + backports-tarfile==1.2.0\n + beautifulsoup4==4.12.3\n + boto==2.49.0\n + certifi==2024.12.14\n + cffi==1.17.1\n + chardet==5.2.0\n + charset-normalizer==3.4.1\n + citeproc-py==0.6.0\n + cryptography==44.0.0\n + datalad==0.12.6 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237)\n + deprecated==1.2.15\n + distro==1.9.0\n + duecredit==0.10.2\n + exifread==3.0.0\n + fasteners==0.19\n + gitdb==4.0.11\n + gitpython==3.1.43\n + httpretty==1.1.4\n + humanize==4.11.0\n + idna==3.10\n + importlib-metadata==8.5.0\n + iso8601==2.1.0\n + jaraco-classes==3.4.0\n + jaraco-context==6.0.1\n + jaraco-functools==4.1.0\n + jeepney==0.8.0\n + jsmin==3.0.1\n + keyring==25.6.0\n + keyrings-alt==5.0.2\n + looseversion==1.3.0\n + lxml==5.3.0\n + more-itertools==10.5.0\n + msgpack==1.1.0\n + multidict==6.1.0\n + mutagen==1.47.0\n + nose==1.3.7\n + patool==1.15.0\n + pillow==11.0.0\n + propcache==0.2.1\n + pycparser==2.22\n + pygithub==2.5.0\n + pyjwt==2.10.1\n + pynacl==1.5.0\n + pyperclip==1.9.0\n + python-xmp-toolkit==2.0.2\n + pyyaml==6.0.2\n + requests==2.32.3\n + requests-ftp==0.3.1\n + secretstorage==3.3.3\n + simplejson==3.19.3\n + smmap==5.0.1\n + soupsieve==2.6\n + tqdm==4.67.1\n + urllib3==1.26.20\n + vcrpy==7.0.0\n + whoosh==2.7.4\n + wrapt==1.17.0\n + yarl==1.18.3\nResolved 99 packages in 151ms\nInstalled 32 packages in 610ms\n + alabaster==0.7.16\n + asv==0.6.4\n + asv-runner==0.2.1\n + babel==2.16.0\n + build==1.2.2.post1\n + distlib==0.3.9\n + docutils==0.21.2\n + filelock==3.16.1\n + gprof2dot==2024.6.6\n + imagesize==1.4.1\n + jinja2==3.1.5\n + json5==0.10.0\n + markupsafe==3.0.2\n + nose-timer==1.0.1\n + platformdirs==4.3.6\n + psutil==6.1.1\n + pygments==2.18.0\n + pympler==1.1\n + pypandoc==1.14\n + pyproject-hooks==1.2.0\n + snowballstemmer==2.2.0\n + sphinx==7.4.7\n + sphinx-rtd-theme==3.0.2\n + sphinxcontrib-applehelp==2.0.0\n + sphinxcontrib-devhelp==2.0.0\n + sphinxcontrib-htmlhelp==2.1.0\n + sphinxcontrib-jquery==4.1\n + sphinxcontrib-jsmath==1.0.1\n + sphinxcontrib-qthelp==2.0.0\n + sphinxcontrib-serializinghtml==2.0.0\n + tabulate==0.9.0\n + virtualenv==20.28.0\n",
    "new_commit_res_code": 1,
    "new_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.9.21, pytest-8.3.4, pluggy-1.5.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237\nplugins: cov-6.0.0\ncollected 17 items\n\nr2e_tests/test_1.py ..EEx.E.EE...EEEE                                    [100%]\n\n==================================== ERRORS ====================================\n_________________ ERROR at setup of test_HTTPDownloader_basic __________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 126\n  @known_failure_githubci_win\n  @with_tree(tree=[('file.dat', 'abc')])\n  @serve_path_via_http\n  def test_HTTPDownloader_basic(toppath, topurl):\nE       fixture 'toppath' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:126\n_____________________ ERROR at setup of test_access_denied _____________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 198\n  @with_tree(tree=[('file.dat', 'abc')])\n  @serve_path_via_http\n  @with_memory_keyring\n  def test_access_denied(toppath, topurl, keyring):\nE       fixture 'toppath' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:198\n_________________________ ERROR at setup of test_mtime _________________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 359\n  @with_tree(tree={'file.dat': '1'})\n  @serve_path_via_http\n  @with_tempfile\n  def test_mtime(path, url, tempfile):\nE       fixture 'url' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:359\n____________ ERROR at setup of test_HTMLFormAuthenticator_httpretty ____________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 433\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  def test_HTMLFormAuthenticator_httpretty(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:433\n___________________ ERROR at setup of test_auth_but_no_cred ____________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 500\n  @with_memory_keyring\n  @with_testsui(responses=['no', 'yes', 'testlogin', 'testpassword'])\n  def test_auth_but_no_cred(keyring):\nE       fixture 'keyring' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:500\n______________________ ERROR at setup of test_scenario_2 _______________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 584\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db(cookies={'example.com': dict(some_site_id='idsomething', expires='Tue, 15 Jan 2013 21:47:38 GMT')})\n  def test_scenario_2(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:584\n_____________ ERROR at setup of test_HTTPBearerTokenAuthenticator ______________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 657\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  def test_HTTPBearerTokenAuthenticator(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:657\n______________ ERROR at setup of test_HTTPLorisTokenAuthenticator ______________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 699\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  def test_HTTPLorisTokenAuthenticator(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:699\n_____________________ ERROR at setup of test_lorisadapter ______________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 736\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  @with_memory_keyring\n  @with_testsui(responses=['yes', 'user'])\n  def test_lorisadapter(d, keyring):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:736\n==================================== PASSES ====================================\n______________________________ test_download_ftp _______________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_url43sakqq2' \n[INFO] Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_url43sakqq2' \n[INFO] Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_url43sakqq2' \n[INFO] Fetching 'ftp://ftp.gnu.org/README' \n[INFO] Fetching 'ftp://ftp.gnu.org/README' \n[INFO] Fetching 'ftp://ftp.gnu.org/README' \n------------------------------ Captured log call -------------------------------\nINFO     datalad.downloaders:base.py:502 Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_url43sakqq2'\nINFO     datalad.downloaders:base.py:502 Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_url43sakqq2'\nINFO     datalad.downloaders:base.py:502 Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_url43sakqq2'\nINFO     datalad.downloaders:base.py:610 Fetching 'ftp://ftp.gnu.org/README'\nINFO     datalad.downloaders:base.py:610 Fetching 'ftp://ftp.gnu.org/README'\nINFO     datalad.downloaders:base.py:610 Fetching 'ftp://ftp.gnu.org/README'\n_________________________ test_authfail404_interactive _________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail40459_07gfq' \n[INFO] http session: Authenticating into session for http://example.com/crap.txt \n[INFO] http session: Authenticating into session for http://example.com/crap.txt \n------------------------------ Captured log call -------------------------------\nINFO     datalad.downloaders:base.py:502 Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail40459_07gfq'\nINFO     datalad.http:http.py:150 http session: Authenticating into session for http://example.com/crap.txt\nINFO     datalad.http:http.py:150 http session: Authenticating into session for http://example.com/crap.txt\n_______________________ test_authfail404_noninteractive ________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail4041n3mes1a' \n[INFO] http session: Authenticating into session for http://example.com/crap.txt \n[ERROR] Interface is non interactive, so we are reraising: Authentication to http://example.com/crap.txt failed: returned output which matches regular expression Failed [http.py:check_for_auth_failure:221] \n------------------------------ Captured log call -------------------------------\nINFO     datalad.downloaders:base.py:502 Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail4041n3mes1a'\nINFO     datalad.http:http.py:150 http session: Authenticating into session for http://example.com/crap.txt\nERROR    datalad.downloaders:base.py:235 Interface is non interactive, so we are reraising: Authentication to http://example.com/crap.txt failed: returned output which matches regular expression Failed [http.py:check_for_auth_failure:221]\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_docstring\nPASSED r2e_tests/test_1.py::test_process_www_authenticate\nPASSED r2e_tests/test_1.py::test_download_ftp\nPASSED r2e_tests/test_1.py::test_get_status_from_headers\nPASSED r2e_tests/test_1.py::test_authfail404_interactive\nPASSED r2e_tests/test_1.py::test_authfail404_noninteractive\nPASSED r2e_tests/test_1.py::test_auth_bytes_content\nXFAIL r2e_tests/test_1.py::test_authenticate_external_portals - reason: [NOTRUN] yield tests were removed in pytest 4.0 - test_authenticate_external_portals will be ignored\nERROR r2e_tests/test_1.py::test_HTTPDownloader_basic\nERROR r2e_tests/test_1.py::test_access_denied\nERROR r2e_tests/test_1.py::test_mtime\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty\nERROR r2e_tests/test_1.py::test_auth_but_no_cred\nERROR r2e_tests/test_1.py::test_scenario_2\nERROR r2e_tests/test_1.py::test_HTTPBearerTokenAuthenticator\nERROR r2e_tests/test_1.py::test_HTTPLorisTokenAuthenticator\nERROR r2e_tests/test_1.py::test_lorisadapter\n==================== 7 passed, 1 xfailed, 9 errors in 6.40s ====================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.9.21, pytest-8.3.4, pluggy-1.5.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237\nplugins: cov-6.0.0\ncollected 17 items\n\nr2e_tests/test_1.py ..EEx.E.EE..FEEEE                                    [100%]\n\n==================================== ERRORS ====================================\n_________________ ERROR at setup of test_HTTPDownloader_basic __________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 126\n  @known_failure_githubci_win\n  @with_tree(tree=[('file.dat', 'abc')])\n  @serve_path_via_http\n  def test_HTTPDownloader_basic(toppath, topurl):\nE       fixture 'toppath' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:126\n_____________________ ERROR at setup of test_access_denied _____________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 198\n  @with_tree(tree=[('file.dat', 'abc')])\n  @serve_path_via_http\n  @with_memory_keyring\n  def test_access_denied(toppath, topurl, keyring):\nE       fixture 'toppath' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:198\n_________________________ ERROR at setup of test_mtime _________________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 359\n  @with_tree(tree={'file.dat': '1'})\n  @serve_path_via_http\n  @with_tempfile\n  def test_mtime(path, url, tempfile):\nE       fixture 'url' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:359\n____________ ERROR at setup of test_HTMLFormAuthenticator_httpretty ____________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 433\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  def test_HTMLFormAuthenticator_httpretty(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:433\n___________________ ERROR at setup of test_auth_but_no_cred ____________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 500\n  @with_memory_keyring\n  @with_testsui(responses=['no', 'yes', 'testlogin', 'testpassword'])\n  def test_auth_but_no_cred(keyring):\nE       fixture 'keyring' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:500\n______________________ ERROR at setup of test_scenario_2 _______________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 584\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db(cookies={'example.com': dict(some_site_id='idsomething', expires='Tue, 15 Jan 2013 21:47:38 GMT')})\n  def test_scenario_2(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:584\n_____________ ERROR at setup of test_HTTPBearerTokenAuthenticator ______________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 657\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  def test_HTTPBearerTokenAuthenticator(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:657\n______________ ERROR at setup of test_HTTPLorisTokenAuthenticator ______________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 699\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  def test_HTTPLorisTokenAuthenticator(d):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:699\n_____________________ ERROR at setup of test_lorisadapter ______________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py, line 736\n  @skip_if(not httpretty, \"no httpretty\")\n  @without_http_proxy\n  @httpretty.activate\n  @with_tempfile(mkdir=True)\n  @with_fake_cookies_db\n  @with_memory_keyring\n  @with_testsui(responses=['yes', 'user'])\n  def test_lorisadapter(d, keyring):\nE       fixture 'd' not found\n>       available fixtures: cache, capfd, capfdbinary, caplog, capsys, capsysbinary, cov, doctest_namespace, monkeypatch, no_cover, path, pytestconfig, record_property, record_testsuite_property, record_xml_attribute, recwarn, setup_git_config, tmp_path, tmp_path_factory, tmpdir, tmpdir_factory\n>       use 'pytest --fixtures [testpath]' for help on them.\n\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_0174e66d85d0fa42d25f1231039a7889ef964237/r2e_tests/test_1.py:736\n=================================== FAILURES ===================================\n___________________________ test_auth_bytes_content ____________________________\n\n    def test_auth_bytes_content():\n        # Our regexes are strings, but we can get content in bytes:\n        # I am not sure yet either we shouldn't just skip then testing for regex,\n        # but we definetely should not crash.\n        authenticator = HTTPBaseAuthenticator(failure_re=\"Failed\")\n>       authenticator.check_for_auth_failure(b\"bytes\")\n\nr2e_tests/test_1.py:567: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \ndatalad/downloaders/http.py:213: in check_for_auth_failure\n    if re.search(failure_re, content):\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n\npattern = 'Failed', string = b'bytes', flags = 0\n\n    def search(pattern, string, flags=0):\n        \"\"\"Scan through string looking for a match to the pattern, returning\n        a Match object, or None if no match was found.\"\"\"\n>       return _compile(pattern, flags).search(string)\nE       TypeError: cannot use a string pattern on a bytes-like object\n\n../../../../.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/re.py:201: TypeError\n==================================== PASSES ====================================\n______________________________ test_download_ftp _______________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_urlp8iyvade' \n[INFO] Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_urlp8iyvade' \n[INFO] Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_urlp8iyvade' \n[INFO] Fetching 'ftp://ftp.gnu.org/README' \n[INFO] Fetching 'ftp://ftp.gnu.org/README' \n[INFO] Fetching 'ftp://ftp.gnu.org/README' \n------------------------------ Captured log call -------------------------------\nINFO     datalad.downloaders:base.py:502 Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_urlp8iyvade'\nINFO     datalad.downloaders:base.py:502 Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_urlp8iyvade'\nINFO     datalad.downloaders:base.py:502 Downloading 'ftp://ftp.gnu.org/README' into '/tmp/datalad_temp_check_download_external_urlp8iyvade'\nINFO     datalad.downloaders:base.py:610 Fetching 'ftp://ftp.gnu.org/README'\nINFO     datalad.downloaders:base.py:610 Fetching 'ftp://ftp.gnu.org/README'\nINFO     datalad.downloaders:base.py:610 Fetching 'ftp://ftp.gnu.org/README'\n_________________________ test_authfail404_interactive _________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail404pshiblz_' \n[INFO] http session: Authenticating into session for http://example.com/crap.txt \n[INFO] http session: Authenticating into session for http://example.com/crap.txt \n------------------------------ Captured log call -------------------------------\nINFO     datalad.downloaders:base.py:502 Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail404pshiblz_'\nINFO     datalad.http:http.py:146 http session: Authenticating into session for http://example.com/crap.txt\nINFO     datalad.http:http.py:146 http session: Authenticating into session for http://example.com/crap.txt\n_______________________ test_authfail404_noninteractive ________________________\n----------------------------- Captured stderr call -----------------------------\n[INFO] Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail404tr193ac6' \n[INFO] http session: Authenticating into session for http://example.com/crap.txt \n[ERROR] Interface is non interactive, so we are reraising: Authentication to http://example.com/crap.txt failed: returned output which matches regular expression Failed [http.py:check_for_auth_failure:214] \n------------------------------ Captured log call -------------------------------\nINFO     datalad.downloaders:base.py:502 Downloading 'http://example.com/crap.txt' into '/tmp/datalad_temp_check_httpretty_authfail404tr193ac6'\nINFO     datalad.http:http.py:146 http session: Authenticating into session for http://example.com/crap.txt\nERROR    datalad.downloaders:base.py:235 Interface is non interactive, so we are reraising: Authentication to http://example.com/crap.txt failed: returned output which matches regular expression Failed [http.py:check_for_auth_failure:214]\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_docstring\nPASSED r2e_tests/test_1.py::test_process_www_authenticate\nPASSED r2e_tests/test_1.py::test_download_ftp\nPASSED r2e_tests/test_1.py::test_get_status_from_headers\nPASSED r2e_tests/test_1.py::test_authfail404_interactive\nPASSED r2e_tests/test_1.py::test_authfail404_noninteractive\nXFAIL r2e_tests/test_1.py::test_authenticate_external_portals - reason: [NOTRUN] yield tests were removed in pytest 4.0 - test_authenticate_external_portals will be ignored\nERROR r2e_tests/test_1.py::test_HTTPDownloader_basic\nERROR r2e_tests/test_1.py::test_access_denied\nERROR r2e_tests/test_1.py::test_mtime\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty\nERROR r2e_tests/test_1.py::test_auth_but_no_cred\nERROR r2e_tests/test_1.py::test_scenario_2\nERROR r2e_tests/test_1.py::test_HTTPBearerTokenAuthenticator\nERROR r2e_tests/test_1.py::test_HTTPLorisTokenAuthenticator\nERROR r2e_tests/test_1.py::test_lorisadapter\nFAILED r2e_tests/test_1.py::test_auth_bytes_content - TypeError: cannot use a...\n=============== 1 failed, 6 passed, 1 xfailed, 9 errors in 4.69s ===============\n",
    "old_commit_res_stderr": ""
}
datalad/downloaders/http.py datalad/downloaders/tests/test_http.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** TypeError When Matching Bytes Content with String Regex

**Description:**
When processing authentication failure responses, if the content is received as bytes, using a string-based regular expression leads to a `TypeError`. This occurs because the regex pattern is not compatible with the bytes content, causing the authentication failure check to crash instead of handling the failure gracefully.

**Example Code:**
```python
authenticator = HTTPBaseAuthenticator(failure_re="Failed")
authenticator.check_for_auth_failure(b"bytes")
```

**Expected Behavior:**
The `check_for_auth_failure` method should handle bytes content by using a bytes-compatible regular expression. Instead of raising an error, it should correctly determine whether the content matches the failure pattern and respond appropriately.

**Actual Behavior:**
A `TypeError` is raised with the message: 
```
TypeError: cannot use a string pattern on a bytes-like object
```
This prevents the authentication failure from being handled properly and causes the application to crash unexpectedly.
[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### datalad/downloaders/http.py
# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""Provide access to stuff (html, data files) via HTTP and HTTPS

"""
import re
import requests
import requests.auth
# at some point was trying to be too specific about which exceptions to
# catch for a retry of a download.
# from urllib3.exceptions import MaxRetryError, NewConnectionError

import io
from time import sleep

from ..utils import assure_list_from_str, assure_dict_from_str
from ..dochelpers import borrowkwargs

from ..ui import ui
from ..utils import auto_repr
from ..dochelpers import exc_str
from ..support.network import get_url_filename
from ..support.network import get_response_disposition_filename
from ..support.network import rfc2822_to_epoch
from ..support.cookies import cookies_db
from ..support.status import FileStatus
from ..support.exceptions import (
    DownloadError,
    AccessDeniedError,
    AccessFailedError,
    UnhandledRedirectError,
)

from .base import Authenticator
from .base import BaseDownloader, DownloaderSession

from logging import getLogger
from ..log import LoggerHelper
lgr = getLogger('datalad.http')

try:
    import requests_ftp
    requests_ftp.monkeypatch_session()
except ImportError as e:
    lgr.debug("Failed to import requests_ftp, thus no ftp support: %s" % exc_str(e))
    pass

if lgr.getEffectiveLevel() <= 1:
    # Let's also enable requests etc debugging

    # These two lines enable debugging at httplib level (requests->urllib3->http.client)
    # You will see the REQUEST, including HEADERS and DATA, and RESPONSE with HEADERS but without DATA.
    # The only thing missing will be the response.body which is not logged.
    import http.client
    # TODO: nohow wrapped with logging, plain prints (heh heh), so formatting will not be consistent
    http.client.HTTPConnection.debuglevel = 1

    # for requests we can define logging properly
    requests_log = LoggerHelper(logtarget="requests.packages.urllib3").get_initialized_logger()
    requests_log.setLevel(lgr.getEffectiveLevel())
    requests_log.propagate = True

__docformat__ = 'restructuredtext'


def process_www_authenticate(v):
    if not v:
        return []
    supported_type = v.split(' ')[0].lower()
    our_type = {
        'basic': 'http_basic_auth',
        'digest': 'http_digest_auth',
    }.get(supported_type)
    return [our_type] if our_type else []


def check_response_status(response, err_prefix="", session=None):
    """Check if response's status_code signals problem with authentication etc

    ATM succeeds only if response code was 200
    """
    if not err_prefix:
        err_prefix = "Access to %s has failed: " % response.url
    # 401 would be for digest authentication mechanism, or if we first ask which mechanisms are
    # supported.... must be linked into the logic if we decide to automagically detect which
    # mechanism or to give more sensible error message
    err_msg = err_prefix + "status code %d" % response.status_code
    if response.status_code in {404}:
        # It could have been that form_url is wrong, so let's just say that
        # TODO: actually may be that is where we could use tagid and actually determine the form submission url
        raise DownloadError(err_prefix + "not found")
    elif 400 <= response.status_code < 500:
        raise AccessDeniedError(
            err_msg,
            supported_types=process_www_authenticate(
                response.headers.get('WWW-Authenticate')))
    elif response.status_code in {200}:
        pass
    elif response.status_code in {301, 302, 307}:
        # TODO: apparently tests do not excercise this one yet
        if session is None:
            raise AccessFailedError(err_msg + " no session was provided")
        redirs = list(session.resolve_redirects(response, response.request))
        if len(redirs) > 1:
            lgr.warning("Multiple redirects aren't supported yet.  Taking first")
        elif len(redirs) == 0:
            raise AccessFailedError("No redirects were resolved")
        raise UnhandledRedirectError(err_msg, url=redirs[0].url)
    else:
        raise AccessFailedError(err_msg)


@auto_repr
class HTTPBaseAuthenticator(Authenticator):
    """Base class for html_form and http_auth authenticators
    """
    def __init__(self, url=None, failure_re=None, success_re=None,
                 session_cookies=None, **kwargs):
        """
        Parameters
        ----------
        url : str, optional
          URL where to find the form/login to authenticate.  If not provided, an original query url
          which will be provided to the __call__ of the authenticator will be used
        failure_re : str or list of str, optional
        success_re : str or list of str, optional
          Regular expressions to determine if login has failed or succeeded.
          TODO: we might condition when it gets ran
        session_cookies : str or list of str, optional
          Session cookies to store (besides auth response cookies)
        """
        super(HTTPBaseAuthenticator, self).__init__(**kwargs)
        self.url = url
        self.failure_re = assure_list_from_str(failure_re)
        self.success_re = assure_list_from_str(success_re)
        self.session_cookies = assure_list_from_str(session_cookies)

    def authenticate(self, url, credential, session, update=False):
        # we should use specified URL for this authentication first
        lgr.info("http session: Authenticating into session for %s", url)
        post_url = self.url if self.url else url
        credentials = credential()

        # The whole thing relies on server first spitting out 401
        # and client GETing again with 'Authentication:' header
        # So we need custom handling for those, while keeping track not
        # of cookies per se, but of 'Authentication:' header which is
        # to be used in subsequent GETs
        response = self._post_credential(credentials, post_url, session)
        if response is None:
            # authentication did not involve any interaction, nothing to
            # check at this point
            return

        # Handle responses if there was initial authentication exchange,
        # e.g. posting to a form and getting a cookie etc
        err_prefix = "Authentication to %s failed: " % post_url
        try:
            check_response_status(response, err_prefix, session=session)
        except DownloadError:
            # It might have happened that the return code was 'incorrect'
            # and we did get some feedback, which we could analyze to
            # figure out actual problem.  E.g. in case of nersc of crcns
            # it returns 404 (not found) with text in the html
            if response is not None and response.text:
                self.check_for_auth_failure(response.text, err_prefix)
            raise

        response_text = response.text
        self.check_for_auth_failure(response_text, err_prefix)

        if self.success_re:
            # the one which must be used to verify success
            # verify that we actually logged in
            for success_re in self.success_re:
                if not re.search(success_re, response_text):
                    raise AccessDeniedError(
                        err_prefix + " returned output did not match 'success' regular expression %s" % success_re
                    )

        cookies_dict = {}
        if response.cookies:
            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)
        if self.session_cookies:
            # any session cookies to store
            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})

        if cookies_dict:
            if (url in cookies_db) and update:
                cookies_db[url].update(cookies_dict)
            else:
                cookies_db[url] = cookies_dict
            # assign cookies for this session
            for c, v in cookies_dict.items():
                if c not in session.cookies or session.cookies[c] != v:
                    session.cookies[c] = v  # .update(cookies_dict)

        return response

    def _post_credential(self, credentials, post_url, session):
        raise NotImplementedError("Must be implemented in subclass")

    def check_for_auth_failure(self, content, err_prefix=""):
        if self.failure_re:
            # verify that we actually logged in
            for failure_re in self.failure_re:
                if re.search(failure_re, content):
                    raise AccessDeniedError(
                        err_prefix + "returned output which matches regular expression %s" % failure_re
                    )


@auto_repr
class HTMLFormAuthenticator(HTTPBaseAuthenticator):
    """Authenticate by opening a session via POSTing to HTML form
    """

    def __init__(self, fields, tagid=None, **kwargs):
        """

        Example specification in the .ini config file
        [provider:crcns]
        ...
        credential = crcns ; is not given to authenticator as is
        authentication_type = html_form
        # TODO: may be rename into post_url
        html_form_url = https://crcns.org/login_form
        # probably not needed actually since form_url
        # html_form_tagid = login_form
        html_form_fields = __ac_name={user}
                   __ac_password={password}
                   submit=Log in
                   form.submitted=1
                   js_enabled=0
                   cookies_enabled=
        html_form_failure_re = (Login failed|Please log in)
        html_form_success_re = You are now logged in

        Parameters
        ----------
        fields : str or dict
          String or a dictionary, which will be used (along with credential) information
          to feed into the form
        tagid : str, optional
          id of the HTML <form> in the document to use. If None, and page contains a single form,
          that one will be used.  If multiple forms -- error will be raise
        **kwargs : dict, optional
          Passed to super class HTTPBaseAuthenticator
        """
        super(HTMLFormAuthenticator, self).__init__(**kwargs)
        self.fields = assure_dict_from_str(fields)
        self.tagid = tagid

    def _post_credential(self, credentials, post_url, session):
        post_fields = {
            k: v.format(**credentials)
            for k, v in self.fields.items()
        }

        response = session.post(post_url, data=post_fields)
        lgr.debug("Posted to %s fields %s, got response %s with headers %s",
                  post_url, list(post_fields.keys()), response,
                  list(response.headers.keys()))
        return response


@auto_repr
class HTTPRequestsAuthenticator(HTTPBaseAuthenticator):
    """Base class for various authenticators using requests pre-crafted ones
    """

    REQUESTS_AUTHENTICATOR = None
    REQUESTS_FIELDS = ('user', 'password')

    def __init__(self, **kwargs):
        # so we have __init__ solely for a custom docstring
        super(HTTPRequestsAuthenticator, self).__init__(**kwargs)

    def _post_credential(self, credentials, post_url, session):
        authenticator = self.REQUESTS_AUTHENTICATOR(
            *[credentials[f] for f in self.REQUESTS_FIELDS])
        session.auth = authenticator


@auto_repr
class HTTPBasicAuthAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via basic HTTP authentication

    Example specification in the .ini config file
    [provider:hcp-db]
    ...
    credential = hcp-db
    authentication_type = http_auth

    Parameters
    ----------
    **kwargs : dict, optional
      Passed to super class HTTPBaseAuthenticator
    """

    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth


@auto_repr
class HTTPAuthAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via Basic authentication to some other post url

    TODO:  actually this is some remnants which might later were RFed
    into the form authenticator since otherwise they make little sense
    """

    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth

    def _post_credential(self, credentials, post_url, session):
        authenticator = self.REQUESTS_AUTHENTICATOR(
            *[credentials[f] for f in self.REQUESTS_FIELDS])
        session.auth = authenticator
        response = session.post(post_url, data={},
                                auth=authenticator)
        return response


@auto_repr
class HTTPDigestAuthAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via HTTP digest authentication
    """

    REQUESTS_AUTHENTICATOR = requests.auth.HTTPDigestAuth


@auto_repr
class HTTPBearerTokenAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via HTTP Authorization header
    """

    DEFAULT_CREDENTIAL_TYPE = 'token'

    def __init__(self, **kwargs):
        # so we have __init__ solely for a custom docstring
        super(HTTPBearerTokenAuthenticator, self).__init__(**kwargs)

    def _post_credential(self, credentials, post_url, session):
        # we do not need to post anything, just inject token into the session
        session.headers['Authorization'] = "Bearer %s" % credentials['token']


@auto_repr
class HTTPDownloaderSession(DownloaderSession):
    def __init__(self, size=None, filename=None,  url=None, headers=None,
                 response=None, chunk_size=1024 ** 2):
        super(HTTPDownloaderSession, self).__init__(
            size=size, filename=filename, url=url, headers=headers,
        )
        self.chunk_size = chunk_size
        self.response = response

    def download(self, f=None, pbar=None, size=None):
        response = self.response
        # content_gzipped = 'gzip' in response.headers.get('content-encoding', '').split(',')
        # if content_gzipped:
        #     raise NotImplemented("We do not support (yet) gzipped content")
        #     # see https://rationalpie.wordpress.com/2010/06/02/python-streaming-gzip-decompression/
        #     # for ways to implement in python 2 and 3.2's gzip is working better with streams

        total = 0
        return_content = f is None
        if f is None:
            # no file to download to
            # TODO: actually strange since it should have been decoded then...
            f = io.BytesIO()

        # must use .raw to be able avoiding decoding/decompression while downloading
        # to a file
        chunk_size_ = min(self.chunk_size, size) if size is not None else self.chunk_size

        # XXX With requests_ftp BytesIO is provided as response.raw for ftp urls,
        # which has no .stream, so let's do ducktyping and provide our custom stream
        # via BufferedReader for such cases, while maintaining the rest of code
        # intact.  TODO: figure it all out, since doesn't scale for any sizeable download
        if not hasattr(response.raw, 'stream'):
            def _stream():
                buf = io.BufferedReader(response.raw)
                v = True
                while v:
                    v = buf.read(chunk_size_)
                    yield v

            stream = _stream()
        else:
            # XXX TODO -- it must be just a dirty workaround
            # As we discovered with downloads from NITRC all headers come with
            # Content-Encoding: gzip which leads  requests to decode them.  But the point
            # is that ftp links (yoh doesn't think) are gzip compressed for the transfer
            decode_content = not response.url.startswith('ftp://')
            stream = response.raw.stream(chunk_size_, decode_content=decode_content)

        for chunk in stream:
            if chunk:  # filter out keep-alive new chunks
                chunk_len = len(chunk)
                if size is not None and total + chunk_len > size:
                    # trim the download to match target size
                    chunk = chunk[:size - total]
                    chunk_len = len(chunk)
                total += chunk_len
                f.write(chunk)
                try:
                    # TODO: pbar is not robust ATM against > 100% performance ;)
                    if pbar:
                        pbar.update(total)
                except Exception as e:
                    lgr.warning("Failed to update progressbar: %s" % exc_str(e))
                # TEMP
                # see https://github.com/niltonvolpato/python-progressbar/pull/44
                ui.out.flush()
                if size is not None and total >= size:
                    break  # we have done as much as we were asked

        if return_content:
            out = f.getvalue()
            return out


@auto_repr
class HTTPDownloader(BaseDownloader):
    """A stateful downloader to maintain a session to the website
    """

    @borrowkwargs(BaseDownloader)
    def __init__(self, headers={}, **kwargs):
        super(HTTPDownloader, self).__init__(**kwargs)
        self._session = None
        self._headers = headers

    def _establish_session(self, url, allow_old=True):
        """

        Parameters
        ----------
        allow_old: bool, optional
          If a Downloader allows for persistent sessions by some means -- flag
          instructs whether to use previous session, or establish a new one

        Returns
        -------
        bool
          To state if old instance of a session/authentication was used
        """
        if allow_old:
            if self._session:
                lgr.debug("http session: Reusing previous")
                return True  # we used old
            elif url in cookies_db:
                cookie_dict = cookies_db[url]
                lgr.debug("http session: Creating new with old cookies %s", list(cookie_dict.keys()))
                self._session = requests.Session()
                # not sure what happens if cookie is expired (need check to that or exception will prolly get thrown)

                # TODO dict_to_cookiejar doesn't preserve all fields when reversed
                self._session.cookies = requests.utils.cookiejar_from_dict(cookie_dict)
                # TODO cookie could be expired w/ something like (but docs say it should be expired automatically):
                # http://docs.python-requests.org/en/latest/api/#requests.cookies.RequestsCookieJar.clear_expired_cookies
                # self._session.cookies.clear_expired_cookies()
                return True

        lgr.debug("http session: Creating brand new session")
        self._session = requests.Session()
        if self.authenticator:
            self.authenticator.authenticate(url, self.credential, self._session)

        return False

    def get_downloader_session(self, url,
                               allow_redirects=True,
                               use_redirected_url=True,
                               headers=None):
        # TODO: possibly make chunk size adaptive
        # TODO: make it not this ugly -- but at the moment we are testing end-file size
        # while can't know for sure if content was gunziped and whether it all went ok.
        # So safer option -- just request to not have it gzipped
        if headers is None:
            headers = {}
        if 'Accept-Encoding' not in headers:
            headers['Accept-Encoding'] = ''
        # TODO: our tests ATM aren't ready for retries, thus altogether disabled for now
        nretries = 1
        for retry in range(1, nretries+1):
            try:
                response = self._session.get(
                    url, stream=True, allow_redirects=allow_redirects,
                    headers=headers)
            #except (MaxRetryError, NewConnectionError) as exc:
            except Exception as exc:
                # happen to run into those with urls pointing to Amazon,
                # so let's rest and try again
                if retry >= nretries:
                    #import epdb; epdb.serve()
                    raise AccessFailedError(
                        "Failed to establish a new session %d times. "
                        "Last exception was: %s"
                        % (nretries, exc_str(exc)))
                lgr.warning(
                    "Caught exception %s. Will retry %d out of %d times",
                    exc_str(exc), retry+1, nretries)
                sleep(2**retry)

        check_response_status(response, session=self._session)
        headers = response.headers
        lgr.debug("Establishing session for url %s, response headers: %s",
                  url, headers)
        target_size = int(headers.get('Content-Length', '0').strip()) or None
        if use_redirected_url and response.url and response.url != url:
            lgr.debug("URL %s was redirected to %s and thus the later will be used"
                      % (url, response.url))
            url = response.url
        # Consult about filename.  Since we already have headers,
        # should not result in an additional request
        url_filename = get_url_filename(url, headers=headers)

        headers['Url-Filename'] = url_filename
        return HTTPDownloaderSession(
            size=target_size,
            url=response.url,
            filename=url_filename,
            headers=headers,
            response=response
        )

    @classmethod
    def get_status_from_headers(cls, headers):
        """Given HTTP headers, return 'status' record to assess later if link content was changed
        """
        # used for quick checks for HTTP or S3?
        # TODO:  So we will base all statuses on this set? e.g. for Last-Modified if to be
        # mapping from field to its type converter
        HTTP_HEADERS_TO_STATUS = {
            'Content-Length': int,
            'Content-Disposition': str,
            'Last-Modified': rfc2822_to_epoch,
            'Url-Filename': str,
        }
        # Allow for webserver to return them in other casing
        HTTP_HEADERS_TO_STATUS_lower = {s.lower(): (s, t) for s, t in HTTP_HEADERS_TO_STATUS.items()}
        status = {}
        if headers:
            for header_key in headers:
                try:
                    k, t = HTTP_HEADERS_TO_STATUS_lower[header_key.lower()]
                except KeyError:
                    continue
                status[k] = t(headers[header_key])

        # convert to FileStatus
        return FileStatus(
            size=status.get('Content-Length'),
            mtime=status.get('Last-Modified'),
            filename=get_response_disposition_filename(
                status.get('Content-Disposition')) or status.get('Url-Filename')
        )

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
[ISSUE]
**Title:** TypeError When Matching Bytes Content with String Regex

**Description:**
When processing authentication failure responses, if the content is received as bytes, using a string-based regular expression leads to a `TypeError`. This occurs because the regex pattern is not compatible with the bytes content, causing the authentication failure check to crash instead of handling the failure gracefully.

**Example Code:**
```python
authenticator = HTTPBaseAuthenticator(failure_re="Failed")
authenticator.check_for_auth_failure(b"bytes")
```

**Expected Behavior:**
The `check_for_auth_failure` method should handle bytes content by using a bytes-compatible regular expression. Instead of raising an error, it should correctly determine whether the content matches the failure pattern and respond appropriately.

**Actual Behavior:**
A `TypeError` is raised with the message: 
```
TypeError: cannot use a string pattern on a bytes-like object
```
This prevents the authentication failure from being handled properly and causes the application to crash unexpectedly.
[/ISSUE]
{
    "test_docstring": "PASSED",
    "test_process_www_authenticate": "PASSED",
    "test_download_ftp": "PASSED",
    "test_get_status_from_headers": "PASSED",
    "test_authfail404_interactive": "PASSED",
    "test_authfail404_noninteractive": "PASSED",
    "test_auth_bytes_content": "PASSED",
    "test_HTTPDownloader_basic": "ERROR",
    "test_access_denied": "ERROR",
    "test_mtime": "ERROR",
    "test_HTMLFormAuthenticator_httpretty": "ERROR",
    "test_auth_but_no_cred": "ERROR",
    "test_scenario_2": "ERROR",
    "test_HTTPBearerTokenAuthenticator": "ERROR",
    "test_HTTPLorisTokenAuthenticator": "ERROR",
    "test_lorisadapter": "ERROR"
}
diff --git a/datalad/downloaders/http.py b/datalad/downloaders/http.py
index a439ed542..297657001 100644
--- a/datalad/downloaders/http.py
+++ b/datalad/downloaders/http.py
@@ -19,7 +19,11 @@ import requests.auth
 import io
 from time import sleep
 
-from ..utils import assure_list_from_str, assure_dict_from_str
+from ..utils import (
+    assure_list_from_str,
+    assure_dict_from_str,
+    ensure_bytes,
+)
 from ..dochelpers import borrowkwargs
 
 from ..ui import ui
@@ -208,8 +212,11 @@ class HTTPBaseAuthenticator(Authenticator):
 
     def check_for_auth_failure(self, content, err_prefix=""):
         if self.failure_re:
+            content_is_bytes = isinstance(content, bytes)
             # verify that we actually logged in
             for failure_re in self.failure_re:
+                if content_is_bytes:
+                    failure_re = ensure_bytes(failure_re)
                 if re.search(failure_re, content):
                     raise AccessDeniedError(
                         err_prefix + "returned output which matches regular expression %s" % failure_re

{"datalad/downloaders/http.py": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Provide access to stuff (html, data files) via HTTP and HTTPS\n\n\"\"\"\nimport re\nimport requests\nimport requests.auth\n# at some point was trying to be too specific about which exceptions to\n# catch for a retry of a download.\n# from urllib3.exceptions import MaxRetryError, NewConnectionError\n\nimport io\nfrom time import sleep\n\nfrom ..utils import assure_list_from_str, assure_dict_from_str\nfrom ..dochelpers import borrowkwargs\n\nfrom ..ui import ui\nfrom ..utils import auto_repr\nfrom ..dochelpers import exc_str\nfrom ..support.network import get_url_filename\nfrom ..support.network import get_response_disposition_filename\nfrom ..support.network import rfc2822_to_epoch\nfrom ..support.cookies import cookies_db\nfrom ..support.status import FileStatus\nfrom ..support.exceptions import (\n    DownloadError,\n    AccessDeniedError,\n    AccessFailedError,\n    UnhandledRedirectError,\n)\n\nfrom .base import Authenticator\nfrom .base import BaseDownloader, DownloaderSession\n\nfrom logging import getLogger\nfrom ..log import LoggerHelper\nlgr = getLogger('datalad.http')\n\ntry:\n    import requests_ftp\n    requests_ftp.monkeypatch_session()\nexcept ImportError as e:\n    lgr.debug(\"Failed to import requests_ftp, thus no ftp support: %s\" % exc_str(e))\n    pass\n\nif lgr.getEffectiveLevel() <= 1:\n    # Let's also enable requests etc debugging\n\n    # These two lines enable debugging at httplib level (requests->urllib3->http.client)\n    # You will see the REQUEST, including HEADERS and DATA, and RESPONSE with HEADERS but without DATA.\n    # The only thing missing will be the response.body which is not logged.\n    import http.client\n    # TODO: nohow wrapped with logging, plain prints (heh heh), so formatting will not be consistent\n    http.client.HTTPConnection.debuglevel = 1\n\n    # for requests we can define logging properly\n    requests_log = LoggerHelper(logtarget=\"requests.packages.urllib3\").get_initialized_logger()\n    requests_log.setLevel(lgr.getEffectiveLevel())\n    requests_log.propagate = True\n\n__docformat__ = 'restructuredtext'\n\n\ndef process_www_authenticate(v):\n    if not v:\n        return []\n    supported_type = v.split(' ')[0].lower()\n    our_type = {\n        'basic': 'http_basic_auth',\n        'digest': 'http_digest_auth',\n    }.get(supported_type)\n    return [our_type] if our_type else []\n\n\ndef check_response_status(response, err_prefix=\"\", session=None):\n    \"\"\"Check if response's status_code signals problem with authentication etc\n\n    ATM succeeds only if response code was 200\n    \"\"\"\n    if not err_prefix:\n        err_prefix = \"Access to %s has failed: \" % response.url\n    # 401 would be for digest authentication mechanism, or if we first ask which mechanisms are\n    # supported.... must be linked into the logic if we decide to automagically detect which\n    # mechanism or to give more sensible error message\n    err_msg = err_prefix + \"status code %d\" % response.status_code\n    if response.status_code in {404}:\n        # It could have been that form_url is wrong, so let's just say that\n        # TODO: actually may be that is where we could use tagid and actually determine the form submission url\n        raise DownloadError(err_prefix + \"not found\")\n    elif 400 <= response.status_code < 500:\n        raise AccessDeniedError(\n            err_msg,\n            supported_types=process_www_authenticate(\n                response.headers.get('WWW-Authenticate')))\n    elif response.status_code in {200}:\n        pass\n    elif response.status_code in {301, 302, 307}:\n        # TODO: apparently tests do not excercise this one yet\n        if session is None:\n            raise AccessFailedError(err_msg + \" no session was provided\")\n        redirs = list(session.resolve_redirects(response, response.request))\n        if len(redirs) > 1:\n            lgr.warning(\"Multiple redirects aren't supported yet.  Taking first\")\n        elif len(redirs) == 0:\n            raise AccessFailedError(\"No redirects were resolved\")\n        raise UnhandledRedirectError(err_msg, url=redirs[0].url)\n    else:\n        raise AccessFailedError(err_msg)\n\n\n@auto_repr\nclass HTTPBaseAuthenticator(Authenticator):\n    \"\"\"Base class for html_form and http_auth authenticators\n    \"\"\"\n    def __init__(self, url=None, failure_re=None, success_re=None,\n                 session_cookies=None, **kwargs):\n        \"\"\"\n        Parameters\n        ----------\n        url : str, optional\n          URL where to find the form/login to authenticate.  If not provided, an original query url\n          which will be provided to the __call__ of the authenticator will be used\n        failure_re : str or list of str, optional\n        success_re : str or list of str, optional\n          Regular expressions to determine if login has failed or succeeded.\n          TODO: we might condition when it gets ran\n        session_cookies : str or list of str, optional\n          Session cookies to store (besides auth response cookies)\n        \"\"\"\n        super(HTTPBaseAuthenticator, self).__init__(**kwargs)\n        self.url = url\n        self.failure_re = assure_list_from_str(failure_re)\n        self.success_re = assure_list_from_str(success_re)\n        self.session_cookies = assure_list_from_str(session_cookies)\n\n    def authenticate(self, url, credential, session, update=False):\n        # we should use specified URL for this authentication first\n        lgr.info(\"http session: Authenticating into session for %s\", url)\n        post_url = self.url if self.url else url\n        credentials = credential()\n\n        # The whole thing relies on server first spitting out 401\n        # and client GETing again with 'Authentication:' header\n        # So we need custom handling for those, while keeping track not\n        # of cookies per se, but of 'Authentication:' header which is\n        # to be used in subsequent GETs\n        response = self._post_credential(credentials, post_url, session)\n        if response is None:\n            # authentication did not involve any interaction, nothing to\n            # check at this point\n            return\n\n        # Handle responses if there was initial authentication exchange,\n        # e.g. posting to a form and getting a cookie etc\n        err_prefix = \"Authentication to %s failed: \" % post_url\n        try:\n            check_response_status(response, err_prefix, session=session)\n        except DownloadError:\n            # It might have happened that the return code was 'incorrect'\n            # and we did get some feedback, which we could analyze to\n            # figure out actual problem.  E.g. in case of nersc of crcns\n            # it returns 404 (not found) with text in the html\n            if response is not None and response.text:\n                self.check_for_auth_failure(response.text, err_prefix)\n            raise\n\n        response_text = response.text\n        self.check_for_auth_failure(response_text, err_prefix)\n\n        if self.success_re:\n            # the one which must be used to verify success\n            # verify that we actually logged in\n            for success_re in self.success_re:\n                if not re.search(success_re, response_text):\n                    raise AccessDeniedError(\n                        err_prefix + \" returned output did not match 'success' regular expression %s\" % success_re\n                    )\n\n        cookies_dict = {}\n        if response.cookies:\n            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)\n        if self.session_cookies:\n            # any session cookies to store\n            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})\n\n        if cookies_dict:\n            if (url in cookies_db) and update:\n                cookies_db[url].update(cookies_dict)\n            else:\n                cookies_db[url] = cookies_dict\n            # assign cookies for this session\n            for c, v in cookies_dict.items():\n                if c not in session.cookies or session.cookies[c] != v:\n                    session.cookies[c] = v  # .update(cookies_dict)\n\n        return response\n\n    def _post_credential(self, credentials, post_url, session):\n        raise NotImplementedError(\"Must be implemented in subclass\")\n\n    def check_for_auth_failure(self, content, err_prefix=\"\"):\n        if self.failure_re:\n            # verify that we actually logged in\n            for failure_re in self.failure_re:\n                if re.search(failure_re, content):\n                    raise AccessDeniedError(\n                        err_prefix + \"returned output which matches regular expression %s\" % failure_re\n                    )\n\n\n@auto_repr\nclass HTMLFormAuthenticator(HTTPBaseAuthenticator):\n    \"\"\"Authenticate by opening a session via POSTing to HTML form\n    \"\"\"\n\n    def __init__(self, fields, tagid=None, **kwargs):\n        \"\"\"\n\n        Example specification in the .ini config file\n        [provider:crcns]\n        ...\n        credential = crcns ; is not given to authenticator as is\n        authentication_type = html_form\n        # TODO: may be rename into post_url\n        html_form_url = https://crcns.org/login_form\n        # probably not needed actually since form_url\n        # html_form_tagid = login_form\n        html_form_fields = __ac_name={user}\n                   __ac_password={password}\n                   submit=Log in\n                   form.submitted=1\n                   js_enabled=0\n                   cookies_enabled=\n        html_form_failure_re = (Login failed|Please log in)\n        html_form_success_re = You are now logged in\n\n        Parameters\n        ----------\n        fields : str or dict\n          String or a dictionary, which will be used (along with credential) information\n          to feed into the form\n        tagid : str, optional\n          id of the HTML <form> in the document to use. If None, and page contains a single form,\n          that one will be used.  If multiple forms -- error will be raise\n        **kwargs : dict, optional\n          Passed to super class HTTPBaseAuthenticator\n        \"\"\"\n        super(HTMLFormAuthenticator, self).__init__(**kwargs)\n        self.fields = assure_dict_from_str(fields)\n        self.tagid = tagid\n\n    def _post_credential(self, credentials, post_url, session):\n        post_fields = {\n            k: v.format(**credentials)\n            for k, v in self.fields.items()\n        }\n\n        response = session.post(post_url, data=post_fields)\n        lgr.debug(\"Posted to %s fields %s, got response %s with headers %s\",\n                  post_url, list(post_fields.keys()), response,\n                  list(response.headers.keys()))\n        return response\n\n\n@auto_repr\nclass HTTPRequestsAuthenticator(HTTPBaseAuthenticator):\n    \"\"\"Base class for various authenticators using requests pre-crafted ones\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = None\n    REQUESTS_FIELDS = ('user', 'password')\n\n    def __init__(self, **kwargs):\n        # so we have __init__ solely for a custom docstring\n        super(HTTPRequestsAuthenticator, self).__init__(**kwargs)\n\n    def _post_credential(self, credentials, post_url, session):\n        authenticator = self.REQUESTS_AUTHENTICATOR(\n            *[credentials[f] for f in self.REQUESTS_FIELDS])\n        session.auth = authenticator\n\n\n@auto_repr\nclass HTTPBasicAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via basic HTTP authentication\n\n    Example specification in the .ini config file\n    [provider:hcp-db]\n    ...\n    credential = hcp-db\n    authentication_type = http_auth\n\n    Parameters\n    ----------\n    **kwargs : dict, optional\n      Passed to super class HTTPBaseAuthenticator\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth\n\n\n@auto_repr\nclass HTTPAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via Basic authentication to some other post url\n\n    TODO:  actually this is some remnants which might later were RFed\n    into the form authenticator since otherwise they make little sense\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth\n\n    def _post_credential(self, credentials, post_url, session):\n        authenticator = self.REQUESTS_AUTHENTICATOR(\n            *[credentials[f] for f in self.REQUESTS_FIELDS])\n        session.auth = authenticator\n        response = session.post(post_url, data={},\n                                auth=authenticator)\n        return response\n\n\n@auto_repr\nclass HTTPDigestAuthAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via HTTP digest authentication\n    \"\"\"\n\n    REQUESTS_AUTHENTICATOR = requests.auth.HTTPDigestAuth\n\n\n@auto_repr\nclass HTTPBearerTokenAuthenticator(HTTPRequestsAuthenticator):\n    \"\"\"Authenticate via HTTP Authorization header\n    \"\"\"\n\n    DEFAULT_CREDENTIAL_TYPE = 'token'\n\n    def __init__(self, **kwargs):\n        # so we have __init__ solely for a custom docstring\n        super(HTTPBearerTokenAuthenticator, self).__init__(**kwargs)\n\n    def _post_credential(self, credentials, post_url, session):\n        # we do not need to post anything, just inject token into the session\n        session.headers['Authorization'] = \"Bearer %s\" % credentials['token']\n\n\n@auto_repr\nclass HTTPDownloaderSession(DownloaderSession):\n    def __init__(self, size=None, filename=None,  url=None, headers=None,\n                 response=None, chunk_size=1024 ** 2):\n        super(HTTPDownloaderSession, self).__init__(\n            size=size, filename=filename, url=url, headers=headers,\n        )\n        self.chunk_size = chunk_size\n        self.response = response\n\n    def download(self, f=None, pbar=None, size=None):\n        response = self.response\n        # content_gzipped = 'gzip' in response.headers.get('content-encoding', '').split(',')\n        # if content_gzipped:\n        #     raise NotImplemented(\"We do not support (yet) gzipped content\")\n        #     # see https://rationalpie.wordpress.com/2010/06/02/python-streaming-gzip-decompression/\n        #     # for ways to implement in python 2 and 3.2's gzip is working better with streams\n\n        total = 0\n        return_content = f is None\n        if f is None:\n            # no file to download to\n            # TODO: actually strange since it should have been decoded then...\n            f = io.BytesIO()\n\n        # must use .raw to be able avoiding decoding/decompression while downloading\n        # to a file\n        chunk_size_ = min(self.chunk_size, size) if size is not None else self.chunk_size\n\n        # XXX With requests_ftp BytesIO is provided as response.raw for ftp urls,\n        # which has no .stream, so let's do ducktyping and provide our custom stream\n        # via BufferedReader for such cases, while maintaining the rest of code\n        # intact.  TODO: figure it all out, since doesn't scale for any sizeable download\n        if not hasattr(response.raw, 'stream'):\n            def _stream():\n                buf = io.BufferedReader(response.raw)\n                v = True\n                while v:\n                    v = buf.read(chunk_size_)\n                    yield v\n\n            stream = _stream()\n        else:\n            # XXX TODO -- it must be just a dirty workaround\n            # As we discovered with downloads from NITRC all headers come with\n            # Content-Encoding: gzip which leads  requests to decode them.  But the point\n            # is that ftp links (yoh doesn't think) are gzip compressed for the transfer\n            decode_content = not response.url.startswith('ftp://')\n            stream = response.raw.stream(chunk_size_, decode_content=decode_content)\n\n        for chunk in stream:\n            if chunk:  # filter out keep-alive new chunks\n                chunk_len = len(chunk)\n                if size is not None and total + chunk_len > size:\n                    # trim the download to match target size\n                    chunk = chunk[:size - total]\n                    chunk_len = len(chunk)\n                total += chunk_len\n                f.write(chunk)\n                try:\n                    # TODO: pbar is not robust ATM against > 100% performance ;)\n                    if pbar:\n                        pbar.update(total)\n                except Exception as e:\n                    lgr.warning(\"Failed to update progressbar: %s\" % exc_str(e))\n                # TEMP\n                # see https://github.com/niltonvolpato/python-progressbar/pull/44\n                ui.out.flush()\n                if size is not None and total >= size:\n                    break  # we have done as much as we were asked\n\n        if return_content:\n            out = f.getvalue()\n            return out\n\n\n@auto_repr\nclass HTTPDownloader(BaseDownloader):\n    \"\"\"A stateful downloader to maintain a session to the website\n    \"\"\"\n\n    @borrowkwargs(BaseDownloader)\n    def __init__(self, headers={}, **kwargs):\n        super(HTTPDownloader, self).__init__(**kwargs)\n        self._session = None\n        self._headers = headers\n\n    def _establish_session(self, url, allow_old=True):\n        \"\"\"\n\n        Parameters\n        ----------\n        allow_old: bool, optional\n          If a Downloader allows for persistent sessions by some means -- flag\n          instructs whether to use previous session, or establish a new one\n\n        Returns\n        -------\n        bool\n          To state if old instance of a session/authentication was used\n        \"\"\"\n        if allow_old:\n            if self._session:\n                lgr.debug(\"http session: Reusing previous\")\n                return True  # we used old\n            elif url in cookies_db:\n                cookie_dict = cookies_db[url]\n                lgr.debug(\"http session: Creating new with old cookies %s\", list(cookie_dict.keys()))\n                self._session = requests.Session()\n                # not sure what happens if cookie is expired (need check to that or exception will prolly get thrown)\n\n                # TODO dict_to_cookiejar doesn't preserve all fields when reversed\n                self._session.cookies = requests.utils.cookiejar_from_dict(cookie_dict)\n                # TODO cookie could be expired w/ something like (but docs say it should be expired automatically):\n                # http://docs.python-requests.org/en/latest/api/#requests.cookies.RequestsCookieJar.clear_expired_cookies\n                # self._session.cookies.clear_expired_cookies()\n                return True\n\n        lgr.debug(\"http session: Creating brand new session\")\n        self._session = requests.Session()\n        if self.authenticator:\n            self.authenticator.authenticate(url, self.credential, self._session)\n\n        return False\n\n    def get_downloader_session(self, url,\n                               allow_redirects=True,\n                               use_redirected_url=True,\n                               headers=None):\n        # TODO: possibly make chunk size adaptive\n        # TODO: make it not this ugly -- but at the moment we are testing end-file size\n        # while can't know for sure if content was gunziped and whether it all went ok.\n        # So safer option -- just request to not have it gzipped\n        if headers is None:\n            headers = {}\n        if 'Accept-Encoding' not in headers:\n            headers['Accept-Encoding'] = ''\n        # TODO: our tests ATM aren't ready for retries, thus altogether disabled for now\n        nretries = 1\n        for retry in range(1, nretries+1):\n            try:\n                response = self._session.get(\n                    url, stream=True, allow_redirects=allow_redirects,\n                    headers=headers)\n            #except (MaxRetryError, NewConnectionError) as exc:\n            except Exception as exc:\n                # happen to run into those with urls pointing to Amazon,\n                # so let's rest and try again\n                if retry >= nretries:\n                    #import epdb; epdb.serve()\n                    raise AccessFailedError(\n                        \"Failed to establish a new session %d times. \"\n                        \"Last exception was: %s\"\n                        % (nretries, exc_str(exc)))\n                lgr.warning(\n                    \"Caught exception %s. Will retry %d out of %d times\",\n                    exc_str(exc), retry+1, nretries)\n                sleep(2**retry)\n\n        check_response_status(response, session=self._session)\n        headers = response.headers\n        lgr.debug(\"Establishing session for url %s, response headers: %s\",\n                  url, headers)\n        target_size = int(headers.get('Content-Length', '0').strip()) or None\n        if use_redirected_url and response.url and response.url != url:\n            lgr.debug(\"URL %s was redirected to %s and thus the later will be used\"\n                      % (url, response.url))\n            url = response.url\n        # Consult about filename.  Since we already have headers,\n        # should not result in an additional request\n        url_filename = get_url_filename(url, headers=headers)\n\n        headers['Url-Filename'] = url_filename\n        return HTTPDownloaderSession(\n            size=target_size,\n            url=response.url,\n            filename=url_filename,\n            headers=headers,\n            response=response\n        )\n\n    @classmethod\n    def get_status_from_headers(cls, headers):\n        \"\"\"Given HTTP headers, return 'status' record to assess later if link content was changed\n        \"\"\"\n        # used for quick checks for HTTP or S3?\n        # TODO:  So we will base all statuses on this set? e.g. for Last-Modified if to be\n        # mapping from field to its type converter\n        HTTP_HEADERS_TO_STATUS = {\n            'Content-Length': int,\n            'Content-Disposition': str,\n            'Last-Modified': rfc2822_to_epoch,\n            'Url-Filename': str,\n        }\n        # Allow for webserver to return them in other casing\n        HTTP_HEADERS_TO_STATUS_lower = {s.lower(): (s, t) for s, t in HTTP_HEADERS_TO_STATUS.items()}\n        status = {}\n        if headers:\n            for header_key in headers:\n                try:\n                    k, t = HTTP_HEADERS_TO_STATUS_lower[header_key.lower()]\n                except KeyError:\n                    continue\n                status[k] = t(headers[header_key])\n\n        # convert to FileStatus\n        return FileStatus(\n            size=status.get('Content-Length'),\n            mtime=status.get('Last-Modified'),\n            filename=get_response_disposition_filename(\n                status.get('Content-Disposition')) or status.get('Url-Filename')\n        )\n"}
A user will ask you to solve a task. You should first draft your thinking process (inner monologue). Then, generate the solution.

Your response format must follow the template below:
<think>
Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate a correct solution.
</think>
<solution>
Final solution presented to the user.
</solution>
system
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** TypeError When Matching Bytes Content with String Regex

**Description:**
When processing authentication failure responses, if the content is received as bytes, using a string-based regular expression leads to a `TypeError`. This occurs because the regex pattern is not compatible with the bytes content, causing the authentication failure check to crash instead of handling the failure gracefully.

**Example Code:**
```python
authenticator = HTTPBaseAuthenticator(failure_re="Failed")
authenticator.check_for_auth_failure(b"bytes")
```

**Expected Behavior:**
The `check_for_auth_failure` method should handle bytes content by using a bytes-compatible regular expression. Instead of raising an error, it should correctly determine whether the content matches the failure pattern and respond appropriately.

**Actual Behavior:**
A `TypeError` is raised with the message: 
```
TypeError: cannot use a string pattern on a bytes-like object
```
This prevents the authentication failure from being handled properly and causes the application to crash unexpectedly.
[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### datalad/downloaders/http.py
# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""Provide access to stuff (html, data files) via HTTP and HTTPS

"""
import re
import requests
import requests.auth
# at some point was trying to be too specific about which exceptions to
# catch for a retry of a download.
# from urllib3.exceptions import MaxRetryError, NewConnectionError

import io
from time import sleep

from ..utils import assure_list_from_str, assure_dict_from_str
from ..dochelpers import borrowkwargs

from ..ui import ui
from ..utils import auto_repr
from ..dochelpers import exc_str
from ..support.network import get_url_filename
from ..support.network import get_response_disposition_filename
from ..support.network import rfc2822_to_epoch
from ..support.cookies import cookies_db
from ..support.status import FileStatus
from ..support.exceptions import (
    DownloadError,
    AccessDeniedError,
    AccessFailedError,
    UnhandledRedirectError,
)

from .base import Authenticator
from .base import BaseDownloader, DownloaderSession

from logging import getLogger
from ..log import LoggerHelper
lgr = getLogger('datalad.http')

try:
    import requests_ftp
    requests_ftp.monkeypatch_session()
except ImportError as e:
    lgr.debug("Failed to import requests_ftp, thus no ftp support: %s" % exc_str(e))
    pass

if lgr.getEffectiveLevel() <= 1:
    # Let's also enable requests etc debugging

    # These two lines enable debugging at httplib level (requests->urllib3->http.client)
    # You will see the REQUEST, including HEADERS and DATA, and RESPONSE with HEADERS but without DATA.
    # The only thing missing will be the response.body which is not logged.
    import http.client
    # TODO: nohow wrapped with logging, plain prints (heh heh), so formatting will not be consistent
    http.client.HTTPConnection.debuglevel = 1

    # for requests we can define logging properly
    requests_log = LoggerHelper(logtarget="requests.packages.urllib3").get_initialized_logger()
    requests_log.setLevel(lgr.getEffectiveLevel())
    requests_log.propagate = True

__docformat__ = 'restructuredtext'


def process_www_authenticate(v):
    if not v:
        return []
    supported_type = v.split(' ')[0].lower()
    our_type = {
        'basic': 'http_basic_auth',
        'digest': 'http_digest_auth',
    }.get(supported_type)
    return [our_type] if our_type else []


def check_response_status(response, err_prefix="", session=None):
    """Check if response's status_code signals problem with authentication etc

    ATM succeeds only if response code was 200
    """
    if not err_prefix:
        err_prefix = "Access to %s has failed: " % response.url
    # 401 would be for digest authentication mechanism, or if we first ask which mechanisms are
    # supported.... must be linked into the logic if we decide to automagically detect which
    # mechanism or to give more sensible error message
    err_msg = err_prefix + "status code %d" % response.status_code
    if response.status_code in {404}:
        # It could have been that form_url is wrong, so let's just say that
        # TODO: actually may be that is where we could use tagid and actually determine the form submission url
        raise DownloadError(err_prefix + "not found")
    elif 400 <= response.status_code < 500:
        raise AccessDeniedError(
            err_msg,
            supported_types=process_www_authenticate(
                response.headers.get('WWW-Authenticate')))
    elif response.status_code in {200}:
        pass
    elif response.status_code in {301, 302, 307}:
        # TODO: apparently tests do not excercise this one yet
        if session is None:
            raise AccessFailedError(err_msg + " no session was provided")
        redirs = list(session.resolve_redirects(response, response.request))
        if len(redirs) > 1:
            lgr.warning("Multiple redirects aren't supported yet.  Taking first")
        elif len(redirs) == 0:
            raise AccessFailedError("No redirects were resolved")
        raise UnhandledRedirectError(err_msg, url=redirs[0].url)
    else:
        raise AccessFailedError(err_msg)


@auto_repr
class HTTPBaseAuthenticator(Authenticator):
    """Base class for html_form and http_auth authenticators
    """
    def __init__(self, url=None, failure_re=None, success_re=None,
                 session_cookies=None, **kwargs):
        """
        Parameters
        ----------
        url : str, optional
          URL where to find the form/login to authenticate.  If not provided, an original query url
          which will be provided to the __call__ of the authenticator will be used
        failure_re : str or list of str, optional
        success_re : str or list of str, optional
          Regular expressions to determine if login has failed or succeeded.
          TODO: we might condition when it gets ran
        session_cookies : str or list of str, optional
          Session cookies to store (besides auth response cookies)
        """
        super(HTTPBaseAuthenticator, self).__init__(**kwargs)
        self.url = url
        self.failure_re = assure_list_from_str(failure_re)
        self.success_re = assure_list_from_str(success_re)
        self.session_cookies = assure_list_from_str(session_cookies)

    def authenticate(self, url, credential, session, update=False):
        # we should use specified URL for this authentication first
        lgr.info("http session: Authenticating into session for %s", url)
        post_url = self.url if self.url else url
        credentials = credential()

        # The whole thing relies on server first spitting out 401
        # and client GETing again with 'Authentication:' header
        # So we need custom handling for those, while keeping track not
        # of cookies per se, but of 'Authentication:' header which is
        # to be used in subsequent GETs
        response = self._post_credential(credentials, post_url, session)
        if response is None:
            # authentication did not involve any interaction, nothing to
            # check at this point
            return

        # Handle responses if there was initial authentication exchange,
        # e.g. posting to a form and getting a cookie etc
        err_prefix = "Authentication to %s failed: " % post_url
        try:
            check_response_status(response, err_prefix, session=session)
        except DownloadError:
            # It might have happened that the return code was 'incorrect'
            # and we did get some feedback, which we could analyze to
            # figure out actual problem.  E.g. in case of nersc of crcns
            # it returns 404 (not found) with text in the html
            if response is not None and response.text:
                self.check_for_auth_failure(response.text, err_prefix)
            raise

        response_text = response.text
        self.check_for_auth_failure(response_text, err_prefix)

        if self.success_re:
            # the one which must be used to verify success
            # verify that we actually logged in
            for success_re in self.success_re:
                if not re.search(success_re, response_text):
                    raise AccessDeniedError(
                        err_prefix + " returned output did not match 'success' regular expression %s" % success_re
                    )

        cookies_dict = {}
        if response.cookies:
            cookies_dict = requests.utils.dict_from_cookiejar(response.cookies)
        if self.session_cookies:
            # any session cookies to store
            cookies_dict.update({k: session.cookies[k] for k in self.session_cookies})

        if cookies_dict:
            if (url in cookies_db) and update:
                cookies_db[url].update(cookies_dict)
            else:
                cookies_db[url] = cookies_dict
            # assign cookies for this session
            for c, v in cookies_dict.items():
                if c not in session.cookies or session.cookies[c] != v:
                    session.cookies[c] = v  # .update(cookies_dict)

        return response

    def _post_credential(self, credentials, post_url, session):
        raise NotImplementedError("Must be implemented in subclass")

    def check_for_auth_failure(self, content, err_prefix=""):
        if self.failure_re:
            # verify that we actually logged in
            for failure_re in self.failure_re:
                if re.search(failure_re, content):
                    raise AccessDeniedError(
                        err_prefix + "returned output which matches regular expression %s" % failure_re
                    )


@auto_repr
class HTMLFormAuthenticator(HTTPBaseAuthenticator):
    """Authenticate by opening a session via POSTing to HTML form
    """

    def __init__(self, fields, tagid=None, **kwargs):
        """

        Example specification in the .ini config file
        [provider:crcns]
        ...
        credential = crcns ; is not given to authenticator as is
        authentication_type = html_form
        # TODO: may be rename into post_url
        html_form_url = https://crcns.org/login_form
        # probably not needed actually since form_url
        # html_form_tagid = login_form
        html_form_fields = __ac_name={user}
                   __ac_password={password}
                   submit=Log in
                   form.submitted=1
                   js_enabled=0
                   cookies_enabled=
        html_form_failure_re = (Login failed|Please log in)
        html_form_success_re = You are now logged in

        Parameters
        ----------
        fields : str or dict
          String or a dictionary, which will be used (along with credential) information
          to feed into the form
        tagid : str, optional
          id of the HTML <form> in the document to use. If None, and page contains a single form,
          that one will be used.  If multiple forms -- error will be raise
        **kwargs : dict, optional
          Passed to super class HTTPBaseAuthenticator
        """
        super(HTMLFormAuthenticator, self).__init__(**kwargs)
        self.fields = assure_dict_from_str(fields)
        self.tagid = tagid

    def _post_credential(self, credentials, post_url, session):
        post_fields = {
            k: v.format(**credentials)
            for k, v in self.fields.items()
        }

        response = session.post(post_url, data=post_fields)
        lgr.debug("Posted to %s fields %s, got response %s with headers %s",
                  post_url, list(post_fields.keys()), response,
                  list(response.headers.keys()))
        return response


@auto_repr
class HTTPRequestsAuthenticator(HTTPBaseAuthenticator):
    """Base class for various authenticators using requests pre-crafted ones
    """

    REQUESTS_AUTHENTICATOR = None
    REQUESTS_FIELDS = ('user', 'password')

    def __init__(self, **kwargs):
        # so we have __init__ solely for a custom docstring
        super(HTTPRequestsAuthenticator, self).__init__(**kwargs)

    def _post_credential(self, credentials, post_url, session):
        authenticator = self.REQUESTS_AUTHENTICATOR(
            *[credentials[f] for f in self.REQUESTS_FIELDS])
        session.auth = authenticator


@auto_repr
class HTTPBasicAuthAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via basic HTTP authentication

    Example specification in the .ini config file
    [provider:hcp-db]
    ...
    credential = hcp-db
    authentication_type = http_auth

    Parameters
    ----------
    **kwargs : dict, optional
      Passed to super class HTTPBaseAuthenticator
    """

    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth


@auto_repr
class HTTPAuthAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via Basic authentication to some other post url

    TODO:  actually this is some remnants which might later were RFed
    into the form authenticator since otherwise they make little sense
    """

    REQUESTS_AUTHENTICATOR = requests.auth.HTTPBasicAuth

    def _post_credential(self, credentials, post_url, session):
        authenticator = self.REQUESTS_AUTHENTICATOR(
            *[credentials[f] for f in self.REQUESTS_FIELDS])
        session.auth = authenticator
        response = session.post(post_url, data={},
                                auth=authenticator)
        return response


@auto_repr
class HTTPDigestAuthAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via HTTP digest authentication
    """

    REQUESTS_AUTHENTICATOR = requests.auth.HTTPDigestAuth


@auto_repr
class HTTPBearerTokenAuthenticator(HTTPRequestsAuthenticator):
    """Authenticate via HTTP Authorization header
    """

    DEFAULT_CREDENTIAL_TYPE = 'token'

    def __init__(self, **kwargs):
        # so we have __init__ solely for a custom docstring
        super(HTTPBearerTokenAuthenticator, self).__init__(**kwargs)

    def _post_credential(self, credentials, post_url, session):
        # we do not need to post anything, just inject token into the session
        session.headers['Authorization'] = "Bearer %s" % credentials['token']


@auto_repr
class HTTPDownloaderSession(DownloaderSession):
    def __init__(self, size=None, filename=None,  url=None, headers=None,
                 response=None, chunk_size=1024 ** 2):
        super(HTTPDownloaderSession, self).__init__(
            size=size, filename=filename, url=url, headers=headers,
        )
        self.chunk_size = chunk_size
        self.response = response

    def download(self, f=None, pbar=None, size=None):
        response = self.response
        # content_gzipped = 'gzip' in response.headers.get('content-encoding', '').split(',')
        # if content_gzipped:
        #     raise NotImplemented("We do not support (yet) gzipped content")
        #     # see https://rationalpie.wordpress.com/2010/06/02/python-streaming-gzip-decompression/
        #     # for ways to implement in python 2 and 3.2's gzip is working better with streams

        total = 0
        return_content = f is None
        if f is None:
            # no file to download to
            # TODO: actually strange since it should have been decoded then...
            f = io.BytesIO()

        # must use .raw to be able avoiding decoding/decompression while downloading
        # to a file
        chunk_size_ = min(self.chunk_size, size) if size is not None else self.chunk_size

        # XXX With requests_ftp BytesIO is provided as response.raw for ftp urls,
        # which has no .stream, so let's do ducktyping and provide our custom stream
        # via BufferedReader for such cases, while maintaining the rest of code
        # intact.  TODO: figure it all out, since doesn't scale for any sizeable download
        if not hasattr(response.raw, 'stream'):
            def _stream():
                buf = io.BufferedReader(response.raw)
                v = True
                while v:
                    v = buf.read(chunk_size_)
                    yield v

            stream = _stream()
        else:
            # XXX TODO -- it must be just a dirty workaround
            # As we discovered with downloads from NITRC all headers come with
            # Content-Encoding: gzip which leads  requests to decode them.  But the point
            # is that ftp links (yoh doesn't think) are gzip compressed for the transfer
            decode_content = not response.url.startswith('ftp://')
            stream = response.raw.stream(chunk_size_, decode_content=decode_content)

        for chunk in stream:
            if chunk:  # filter out keep-alive new chunks
                chunk_len = len(chunk)
                if size is not None and total + chunk_len > size:
                    # trim the download to match target size
                    chunk = chunk[:size - total]
                    chunk_len = len(chunk)
                total += chunk_len
                f.write(chunk)
                try:
                    # TODO: pbar is not robust ATM against > 100% performance ;)
                    if pbar:
                        pbar.update(total)
                except Exception as e:
                    lgr.warning("Failed to update progressbar: %s" % exc_str(e))
                # TEMP
                # see https://github.com/niltonvolpato/python-progressbar/pull/44
                ui.out.flush()
                if size is not None and total >= size:
                    break  # we have done as much as we were asked

        if return_content:
            out = f.getvalue()
            return out


@auto_repr
class HTTPDownloader(BaseDownloader):
    """A stateful downloader to maintain a session to the website
    """

    @borrowkwargs(BaseDownloader)
    def __init__(self, headers={}, **kwargs):
        super(HTTPDownloader, self).__init__(**kwargs)
        self._session = None
        self._headers = headers

    def _establish_session(self, url, allow_old=True):
        """

        Parameters
        ----------
        allow_old: bool, optional
          If a Downloader allows for persistent sessions by some means -- flag
          instructs whether to use previous session, or establish a new one

        Returns
        -------
        bool
          To state if old instance of a session/authentication was used
        """
        if allow_old:
            if self._session:
                lgr.debug("http session: Reusing previous")
                return True  # we used old
            elif url in cookies_db:
                cookie_dict = cookies_db[url]
                lgr.debug("http session: Creating new with old cookies %s", list(cookie_dict.keys()))
                self._session = requests.Session()
                # not sure what happens if cookie is expired (need check to that or exception will prolly get thrown)

                # TODO dict_to_cookiejar doesn't preserve all fields when reversed
                self._session.cookies = requests.utils.cookiejar_from_dict(cookie_dict)
                # TODO cookie could be expired w/ something like (but docs say it should be expired automatically):
                # http://docs.python-requests.org/en/latest/api/#requests.cookies.RequestsCookieJar.clear_expired_cookies
                # self._session.cookies.clear_expired_cookies()
                return True

        lgr.debug("http session: Creating brand new session")
        self._session = requests.Session()
        if self.authenticator:
            self.authenticator.authenticate(url, self.credential, self._session)

        return False

    def get_downloader_session(self, url,
                               allow_redirects=True,
                               use_redirected_url=True,
                               headers=None):
        # TODO: possibly make chunk size adaptive
        # TODO: make it not this ugly -- but at the moment we are testing end-file size
        # while can't know for sure if content was gunziped and whether it all went ok.
        # So safer option -- just request to not have it gzipped
        if headers is None:
            headers = {}
        if 'Accept-Encoding' not in headers:
            headers['Accept-Encoding'] = ''
        # TODO: our tests ATM aren't ready for retries, thus altogether disabled for now
        nretries = 1
        for retry in range(1, nretries+1):
            try:
                response = self._session.get(
                    url, stream=True, allow_redirects=allow_redirects,
                    headers=headers)
            #except (MaxRetryError, NewConnectionError) as exc:
            except Exception as exc:
                # happen to run into those with urls pointing to Amazon,
                # so let's rest and try again
                if retry >= nretries:
                    #import epdb; epdb.serve()
                    raise AccessFailedError(
                        "Failed to establish a new session %d times. "
                        "Last exception was: %s"
                        % (nretries, exc_str(exc)))
                lgr.warning(
                    "Caught exception %s. Will retry %d out of %d times",
                    exc_str(exc), retry+1, nretries)
                sleep(2**retry)

        check_response_status(response, session=self._session)
        headers = response.headers
        lgr.debug("Establishing session for url %s, response headers: %s",
                  url, headers)
        target_size = int(headers.get('Content-Length', '0').strip()) or None
        if use_redirected_url and response.url and response.url != url:
            lgr.debug("URL %s was redirected to %s and thus the later will be used"
                      % (url, response.url))
            url = response.url
        # Consult about filename.  Since we already have headers,
        # should not result in an additional request
        url_filename = get_url_filename(url, headers=headers)

        headers['Url-Filename'] = url_filename
        return HTTPDownloaderSession(
            size=target_size,
            url=response.url,
            filename=url_filename,
            headers=headers,
            response=response
        )

    @classmethod
    def get_status_from_headers(cls, headers):
        """Given HTTP headers, return 'status' record to assess later if link content was changed
        """
        # used for quick checks for HTTP or S3?
        # TODO:  So we will base all statuses on this set? e.g. for Last-Modified if to be
        # mapping from field to its type converter
        HTTP_HEADERS_TO_STATUS = {
            'Content-Length': int,
            'Content-Disposition': str,
            'Last-Modified': rfc2822_to_epoch,
            'Url-Filename': str,
        }
        # Allow for webserver to return them in other casing
        HTTP_HEADERS_TO_STATUS_lower = {s.lower(): (s, t) for s, t in HTTP_HEADERS_TO_STATUS.items()}
        status = {}
        if headers:
            for header_key in headers:
                try:
                    k, t = HTTP_HEADERS_TO_STATUS_lower[header_key.lower()]
                except KeyError:
                    continue
                status[k] = t(headers[header_key])

        # convert to FileStatus
        return FileStatus(
            size=status.get('Content-Length'),
            mtime=status.get('Last-Modified'),
            filename=get_response_disposition_filename(
                status.get('Content-Disposition')) or status.get('Url-Filename')
        )

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
user