{
    "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\"\"\"Tests for http downloader\"\"\"\n\nimport time\nfrom calendar import timegm\nfrom six import PY3\n\nimport os\nimport six.moves.builtins as __builtin__\nfrom os.path import join as opj\n\nfrom datalad.downloaders.tests.utils import get_test_providers\nfrom ..base import DownloadError\nfrom ..base import IncompleteDownloadError\nfrom ..base import BaseDownloader\nfrom ..credentials import UserPassword\nfrom ..http import HTMLFormAuthenticator\nfrom ..http import HTTPDownloader\nfrom ...support.network import get_url_straight_filename\nfrom ...tests.utils import with_fake_cookies_db\nfrom ...tests.utils import skip_if_no_network\nfrom ...tests.utils import with_testsui\n\n# BTW -- mock_open is not in mock on wheezy (Debian 7.x)\ntry:\n    if PY3:\n        raise ImportError(\"Not yet ready apparently: https://travis-ci.org/datalad/datalad/jobs/111659666\")\n    import httpretty\nexcept ImportError:\n    class NoHTTPPretty(object):\n       __bool__ = __nonzero__ = lambda s: False\n       activate = lambda s, t: t\n    httpretty = NoHTTPPretty()\n\nfrom mock import patch\nfrom ...tests.utils import SkipTest\nfrom ...tests.utils import assert_in\nfrom ...tests.utils import assert_not_in\nfrom ...tests.utils import assert_equal\nfrom ...tests.utils import assert_greater\nfrom ...tests.utils import assert_false\nfrom ...tests.utils import assert_raises\nfrom ...tests.utils import ok_file_has_content\nfrom ...tests.utils import serve_path_via_http, with_tree\nfrom ...tests.utils import swallow_logs\nfrom ...tests.utils import swallow_outputs\nfrom ...tests.utils import with_tempfile\nfrom ...tests.utils import use_cassette\nfrom ...tests.utils import skip_if\nfrom ...tests.utils import without_http_proxy\nfrom ...support.status import FileStatus\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 ;)\ndef fake_open(write_=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(*args, **kwargs):\n        return myfile\n    return myopen\n\n\ndef _raise_IOError(*args, **kwargs):\n    raise IOError(\"Testing here\")\n\n\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    # 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    # 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    with swallow_logs(), \\\n         patch.object(__builtin__, 'open', fake_open(write_=_raise_IOError)):\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_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@use_cassette('test_authenticate_external_portals', record_mode='once')\ndef test_authenticate_external_portals():\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    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\"\ntest_authenticate_external_portals.tags = ['external-portal', 'network']\n\n\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    yield check_download_external_url, \\\n          \"ftp://ftp.gnu.org/README\", \\\n          None, \\\n          \"This is ftp.gnu.org\"\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    assert_equal(\n            HTTPDownloader.get_status_from_headers(headers),\n            FileStatus(size=123, filename='bogus.txt', mtime=1446873816))\n    assert_equal(HTTPDownloader.get_status_from_headers({'content-lengtH': '123'}),\n                 FileStatus(size=123))\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    _fixed_credentials = [\n        {'user': 'testlogin', 'password': 'testpassword'},\n        {'user': 'testlogin2', 'password': 'testpassword2'},\n        {'user': 'testlogin2', 'password': 'testpassword3'}]\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, '')\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@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_testsui(responses=['yes'])  # will request to reentry it\ndef test_HTMLFormAuthenticator_httpretty_authfail404(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, ['failed', '404'])\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_HTMLFormAuthenticator_httpretty_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, '')\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",
            "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\nfrom six import PY3\n\nimport os\nimport six.moves.builtins as __builtin__\nfrom os.path import join as opj\n\nfrom datalad.downloaders.tests.utils import get_test_providers\nfrom ..base import DownloadError\nfrom ..base import IncompleteDownloadError\nfrom ..base import BaseDownloader\nfrom ..credentials import UserPassword\nfrom ..http import HTMLFormAuthenticator\nfrom ..http import HTTPDownloader\nfrom ...support.network import get_url_straight_filename\nfrom ...tests.utils import with_fake_cookies_db\nfrom ...tests.utils import skip_if_no_network\nfrom ...tests.utils import with_testsui\n\n# BTW -- mock_open is not in mock on wheezy (Debian 7.x)\ntry:\n    if PY3:\n        raise ImportError(\"Not yet ready apparently: https://travis-ci.org/datalad/datalad/jobs/111659666\")\n    import httpretty\nexcept ImportError:\n    class NoHTTPPretty(object):\n       __bool__ = __nonzero__ = lambda s: False\n       activate = lambda s, t: t\n    httpretty = NoHTTPPretty()\n\nfrom mock import patch\nfrom ...tests.utils import SkipTest\nfrom ...tests.utils import assert_in\nfrom ...tests.utils import assert_not_in\nfrom ...tests.utils import assert_equal\nfrom ...tests.utils import assert_greater\nfrom ...tests.utils import assert_false\nfrom ...tests.utils import assert_raises\nfrom ...tests.utils import ok_file_has_content\nfrom ...tests.utils import serve_path_via_http, with_tree\nfrom ...tests.utils import swallow_logs\nfrom ...tests.utils import swallow_outputs\nfrom ...tests.utils import with_tempfile\nfrom ...tests.utils import use_cassette\nfrom ...tests.utils import skip_if\nfrom ...tests.utils import without_http_proxy\nfrom ...support.status import FileStatus\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 ;)\ndef fake_open(write_=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(*args, **kwargs):\n        return myfile\n    return myopen\n\n\ndef _raise_IOError(*args, **kwargs):\n    raise IOError(\"Testing here\")\n\n\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    # 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    # 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    with swallow_logs(), \\\n         patch.object(__builtin__, 'open', fake_open(write_=_raise_IOError)):\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_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@use_cassette('test_authenticate_external_portals', record_mode='once')\ndef test_authenticate_external_portals():\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    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\"\ntest_authenticate_external_portals.tags = ['external-portal', 'network']\n\n\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    yield check_download_external_url, \\\n          \"ftp://ftp.gnu.org/README\", \\\n          None, \\\n          \"This is ftp.gnu.org\"\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    assert_equal(\n        HTTPDownloader.get_status_from_headers({\n            'Content-Disposition': 'Attachment;Filename=\"Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip\"',\n        }).filename,\n        'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip')\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    _fixed_credentials = [\n        {'user': 'testlogin', 'password': 'testpassword'},\n        {'user': 'testlogin2', 'password': 'testpassword2'},\n        {'user': 'testlogin2', 'password': 'testpassword3'}]\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, '')\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@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_testsui(responses=['yes'])  # will request to reentry it\ndef test_HTMLFormAuthenticator_httpretty_authfail404(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, ['failed', '404'])\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_HTMLFormAuthenticator_httpretty_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, '')\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",
            "header": {
                "file": {
                    "path": "datalad/downloaders/tests/test_http.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "7320cf16d",
                "new_commit_hash": "01de6706f",
                "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": 246,
                            "length": 12
                        },
                        "new_range": {
                            "start": 246,
                            "length": 21
                        },
                        "section": "def test_get_status_from_headers():"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "        'Last-Modified': 'Sat, 07 Nov 2015 05:23:36 GMT'",
                                "type": "context"
                            },
                            {
                                "content": "    }",
                                "type": "context"
                            },
                            {
                                "content": "    headers['bogus1'] = '123'",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    assert_equal(",
                                "type": "context"
                            },
                            {
                                "content": "            HTTPDownloader.get_status_from_headers(headers),",
                                "type": "context"
                            },
                            {
                                "content": "            FileStatus(size=123, filename='bogus.txt', mtime=1446873816))",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    assert_equal(HTTPDownloader.get_status_from_headers({'content-lengtH': '123'}),",
                                "type": "context"
                            },
                            {
                                "content": "                 FileStatus(size=123))",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    assert_equal(",
                                "type": "added"
                            },
                            {
                                "content": "        HTTPDownloader.get_status_from_headers({",
                                "type": "added"
                            },
                            {
                                "content": "            'Content-Disposition': 'Attachment;Filename=\"Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip\"',",
                                "type": "added"
                            },
                            {
                                "content": "        }).filename,",
                                "type": "added"
                            },
                            {
                                "content": "        'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip')",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "# TODO: test that download fails (even if authentication credentials are right) if form_url",
                                "type": "context"
                            },
                            {
                                "content": "# is wrong!",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/downloaders/tests/test_http.py",
                            "type": "function",
                            "name": "test_get_status_from_headers",
                            "content": "def 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    assert_equal(\n        HTTPDownloader.get_status_from_headers({\n            'Content-Disposition': 'Attachment;Filename=\"Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip\"',\n        }).filename,\n        'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip')",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 240,
                            "end_lineno": 261,
                            "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\nimport calendar\nimport email.utils\nimport gzip\nimport os\nimport re\nimport shutil\nimport time\nimport iso8601\n\nfrom collections import OrderedDict\nfrom os.path import abspath, isabs\n\nfrom six import string_types\nfrom six import iteritems\nfrom six.moves.urllib.request import urlopen, Request\nfrom six.moves.urllib.parse import quote as urlquote, unquote as urlunquote\nfrom six.moves.urllib.parse import urljoin, urlparse, urlsplit, urlunsplit, urlunparse, ParseResult\nfrom six.moves.urllib.parse import parse_qsl\nfrom six.moves.urllib.parse import urlencode\nfrom six.moves.urllib.error import URLError\n\nfrom datalad.utils import on_windows\nfrom datalad.consts import DATASETS_TOPURL\n\nfrom datalad.utils import auto_repr\n# TODO not sure what needs to use `six` here yet\nimport requests\n\nimport logging\nlgr = logging.getLogger('datalad.network')\n\n\ndef get_response_disposition_filename(s):\n    \"\"\"Given a string s as from HTTP Content-Disposition field in the response\n    return possibly present filename if any\n    \"\"\"\n    if not s:\n        return None\n    # If the response has Content-Disposition, try to get filename from it\n    cd = dict(map(\n        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),\n        s.split(';')))\n    if 'filename' in cd:\n        filename = cd['filename'].strip(\"\\\"'\")\n        return filename\n    return None\n\n\ndef get_url_disposition_filename(url, headers=None):\n    \"\"\"Get filename as possibly provided by the server in Content-Disposition\n    \"\"\"\n    if headers is None:\n        request = Request(url)\n        r = retry_urlopen(request)\n        # things are different in requests\n        if 'requests.' in str(r.__class__):\n            headers = r.headers\n        else:\n            headers = r.info()\n    else:\n        r = None\n    try:\n        return get_response_disposition_filename(headers.get('Content-Disposition', ''))\n    finally:\n        if r:\n            r.close()\n\n\ndef get_url_straight_filename(url, strip=[], allowdir=False):\n    \"\"\"Get file/dir name of the last path component of the URL\n\n    Parameters\n    ----------\n    strip: list, optional\n      If provided, listed names will not be considered and their\n      parent directory will be selected\n    allowdir: bool, optional\n      If url points to a \"directory\" (ends with /), empty string\n      would be returned unless allowdir is True, in which case the\n      name of the directory would be returned\n    \"\"\"\n    path = urlunquote(urlsplit(url).path)\n    path_parts = path.split('/')\n\n    if allowdir:\n        # strip empty ones\n        while len(path_parts) > 1 and not path_parts[-1]:\n            path_parts = path_parts[:-1]\n\n    if strip:\n        while path_parts and path_parts[-1] in strip:\n            path_parts = path_parts[:-1]\n\n    if path_parts:\n        return path_parts[-1]\n    else:\n        return None\n\n\ndef get_url_filename(url, headers=None, strip=[]):\n    \"\"\"Get filename from the url, first consulting server about Content-Disposition\n    \"\"\"\n    filename = get_url_disposition_filename(url, headers)\n    if filename:\n        return filename\n    return get_url_straight_filename(url, strip=[])\n\n\ndef get_url_response_stamp(url, response_info):\n    size, mtime = None, None\n    if 'Content-length' in response_info:\n        size = int(response_info['Content-length'])\n    if 'Last-modified' in response_info:\n        mtime = calendar.timegm(email.utils.parsedate(\n            response_info['Last-modified']))\n    return dict(size=size, mtime=mtime, url=url)\n\n\ndef get_tld(url):\n    \"\"\"Return top level domain from a url\n\n    Parameters\n    ----------\n    url : str\n    \"\"\"\n    # maybe use this instead to be safe:  https://pypi.python.org/pypi/tld\n    if not url.strip():\n        raise ValueError(\"Empty URL has no TLD\")\n    rec = urlsplit(url)\n    if not rec.netloc:\n        if not rec.scheme:\n            # There were no scheme provided thus netloc was empty -- must have been a simple 'path like'\n            return url.split('/', 1)[0]\n        else:\n            raise ValueError(\"It seems that only the scheme was provided without the net location/TLD\")\n    return rec.netloc\n\n\nfrom email.utils import parsedate_tz, mktime_tz\ndef rfc2822_to_epoch(datestr):\n    \"\"\"Given rfc2822 date/time format, return seconds since epoch\"\"\"\n    return mktime_tz(parsedate_tz(datestr))\n\n\nimport calendar\ndef iso8601_to_epoch(datestr):\n    \"\"\"Given ISO 8601 date/time format, return in seconds since epoch\n\n    iso8601 is used to parse properly the time zone information, which\n    can't be parsed with standard datetime strptime\n    \"\"\"\n    return calendar.timegm(iso8601.parse_date(datestr).timetuple())\n\n\ndef __urlopen_requests(url, header_vals=None):\n    # XXX Workaround for now for ... broken code\n    if isinstance(url, Request):\n        url = url.get_full_url()\n    return requests.Session().get(url)\n\n\ndef retry_urlopen(url, retries=3):\n    for t in range(retries):\n        try:\n            return __urlopen_requests(url)\n        except URLError as e:\n            lgr.warn(\"Received exception while reading %s: %s\" % (url, e))\n            if t == retries - 1:\n                # if we have reached allowed number of retries -- reraise\n                raise\n\n\ndef is_url_quoted(url):\n    \"\"\"Return either URL looks being already quoted\n    \"\"\"\n    try:\n        url_ = urlunquote(url)\n        return url != url_\n    except:  # problem with unquoting -- then it must be wasn't quoted (correctly)\n        return False\n\n\ndef same_website(url_rec, u_rec):\n    \"\"\"Decide either a link leads to external site\n\n    Parameters\n    ----------\n    url_rec: ParseResult\n      record for original url\n    u_rec: ParseResult\n      record for new url\n    \"\"\"\n    if isinstance(url_rec, string_types):\n        url_rec = urlparse(url_rec)\n    if isinstance(u_rec, string_types):\n        u_rec = urlparse(u_rec)\n    return (url_rec.netloc == u_rec.netloc)\n    # todo: collect more of sample cases.\n    # disabled below check while working on ratholeradio, since links\n    # could go to the parent and that is ok.  Figure out when it was\n    # desired not to go to the parent -- we might need explicit option\n    # and u_rec.path.startswith(url_rec.path)):\n\n\ndef dlurljoin(u_path, url):\n    url_rec = urlparse(url)  # probably duplicating parsing :-/ TODO\n    if url_rec.scheme:\n        # independent full url, so just return it\n        return url\n    if u_path.endswith('/'):  # should here be also a scheme use?\n        if url.startswith('/'):  # jump to the root\n            u_path_rec = urlparse(u_path)\n            return urljoin(urlunparse(\n                (u_path_rec.scheme, u_path_rec.netloc, '', '', '', '')), url)\n        else:\n            return os.path.join(u_path, url)\n    # TODO: recall where all this dirname came from and bring into the test\n    return urljoin(os.path.dirname(u_path) + '/', url)\n\n\n# TODO should it be a node maybe?\nclass SimpleURLStamper(object):\n    \"\"\"Gets a simple stamp about the URL: {url, time, size} of whatever was provided in the header\n    \"\"\"\n    def __init__(self, mode='full'):\n        self.mode = mode\n\n    def __call__(self, url):\n        # Extracted from above madness\n        # TODO: add mode alike to 'relaxed' where we would not\n        # care about content-disposition filename\n        # http://stackoverflow.com/questions/862173/how-to-download-a-file-using-python-in-a-smarter-way\n        request = Request(url)\n\n        # No traffic compression since we do not know how to identify\n        # exactly either it has to be decompressed\n        # request.add_header('Accept-encoding', 'gzip,deflate')\n        #\n        # TODO: think about stamping etc -- we seems to be redoing\n        # what git-annex does for us already... not really\n        r = retry_urlopen(request)\n        try:\n            r_info = r.info()\n            r_stamp = get_url_response_stamp(url, r_info)\n\n            return dict(mtime=r_stamp['mtime'], size=r_stamp['size'], url=url)\n        finally:\n            r.close()\n\n\n# TODO:  make it consistent/clear at what stage % encoding/decoding happens!\n# now it is a mix!\n\n#\n# Useful functionality in requests.models\n#  utils.requote_uri -- quote/unquote cycle to guarantee consistent appearance\n#  RequestEncodingMixin._encode_params -- Will successfully encode parameters when passed as a dict or a list of ...\n#  PreparedRequest().prepare_url(url, params) -- nicely cares about url encodings etc\n#\n\ndef _guess_ri_cls(ri):\n    \"\"\"Factory function which would determine which type of a ri a provided string is\"\"\"\n    # We assume that it is a URL and parse it. Depending on the result\n    # we might decide that it was something else ;)\n    fields = URL._pr_to_fields(urlparse(ri))\n    lgr.log(5, \"Parsed ri %s into fields %s\" % (ri, fields))\n\n    type_ = 'url'\n    # Special treatments\n    # file:///path should stay file:\n    if fields['scheme'] and fields['scheme'] not in {'file'} \\\n            and not fields['hostname']:\n        # dl+archive:... or just for ssh   hostname:path/p1\n        if '+' not in fields['scheme']:\n            type_ = 'ssh'\n            lgr.log(5, \"Assuming ssh style ri, adjusted: %s\" % (fields,))\n\n    if not fields['scheme'] and not fields['hostname']:\n        parts = _split_colon(ri)\n        if fields['path'] and '@' in fields['path'] or len(parts) > 1:\n            # user@host:path/sp1\n            # or host_name: (hence parts check)\n            # TODO: we need a regex to catch those really, parts check is not suff\n            type_ = 'ssh'\n        elif ri.startswith('//'):\n            # e.g. // or ///path\n            type_ = 'datalad'\n        else:\n            type_ = 'file'\n\n    if not fields['scheme'] and fields['hostname']:\n        # e.g. //a/path\n        type_ = 'datalad'\n\n    TYPES = {\n        'url': URL,\n        'ssh':  SSHRI,\n        'file': PathRI,\n        'datalad': DataLadRI\n    }\n    cls = TYPES[type_]\n    # just parse the ri according to regex matchint ssh \"ri\" specs\n    lgr.log(5, \"Detected %s ri\" % type_)\n    return cls\n\n\nclass RI(object):\n    \"\"\"Resource Identifier - base class and a factory for URL, SSHRI, etc\n\n    Intended to be a R/O object (i.e. no fields should be changed in-place).\n    Subclasses define specific collections of fields they care about in _FIELDS\n    class variable.\n    The idea is that this class should help to break apart a URL, while being\n    able to rebuild itself into a string representation for reuse\n\n    `RI` could be used as factory, whenever type of the resource is unknown and\n    must be guessed from the string representation.  One of the subclasses will be\n    provided as output, e.g.\n\n    >>> RI('http://example.com')\n    URL(hostname='example.com', scheme='http')\n    >>> RI('example.com:path')\n    SSHRI(hostname='example.com', path='path')\n    \"\"\"\n\n    # All of the subclasses will provide path\n    _FIELDS = (\n        'path',\n    )\n\n    __slots__ = _FIELDS + ('_fields', '_str')\n\n    def __new__(cls, ri=None, **kwargs):\n        \"\"\"Used as a possible factory for known RI types\n\n        Returns\n        -------\n        RI\n           uninitialized RI object of appropriate class with _str\n           set to string representation if was provided\n\n        \"\"\"\n        if cls is RI and ri is not None:\n            # RI class was used as a factory\n            cls = _guess_ri_cls(ri)\n\n        ri_obj = super(RI, cls).__new__(cls)\n        # Store internally original str\n        ri_obj._str = ri\n        return ri_obj\n\n    def __init__(self, ri=None, **fields):\n        \"\"\"\n        Parameters\n        ----------\n        ri: str, optional\n          String version of a resource specific for this class.  If you would like\n          a type of the resource be deduced, use RI(ri)\n        **fields: dict, optional\n          The values for the fields defined in _FIELDS class variable.\n        \"\"\"\n        if ri and (bool(ri) == bool(fields)):\n            raise ValueError(\n                \"Specify either ri or breakdown from the fields, not both. \"\n                \"Got ri=%r, fields=%r\" % (ri, fields))\n\n        self._fields = self._get_blank_fields()\n        if ri is not None:\n            fields = self._str_to_fields(ri)\n        self._set_from_fields(**fields)\n\n        # If was initialized from a string representation\n        if self._str is not None:\n            # well -- some ris might not unparse identically back\n            # strictly speaking, but let's assume they do\n            ri_ = self.as_str()\n            if ri != ri_:\n                lgr.warning(\"Parsed version of %s %r differs from original %r\",\n                            self.__class__.__name__, ri_, ri)\n\n    @classmethod\n    def _get_blank_fields(cls, **fields):\n        return OrderedDict(((f, fields.get(f, '')) for f in cls._FIELDS))\n\n    @property\n    def fields(self):\n        \"\"\"Returns shallow copy of fields to ease manipulations\"\"\"\n        return self._fields.copy()\n\n    def __repr__(self):\n        # since auto_repr doesn't support \"non-0\" values atm\n        return \"%s(%s)\" % (\n            self.__class__.__name__,\n            \", \".join([\"%s=%r\" % (k, v)\n                       for k, v in sorted(self._fields.items())\n                       if v]))\n\n    # Lazily evaluated if _str was not set\n    def __str__(self):\n        if self._str is None:\n            self._str = self.as_str()\n        return self._str\n\n    @classmethod\n    def from_str(cls, ri_str):\n        obj = cls(**cls._str_to_fields(ri_str))\n        obj._str = ri_str\n        return obj\n\n    @property\n    def localpath(self):\n        # by default RIs point to remote locations\n        raise ValueError(\"%s points to remote location\" % self)\n\n    # Apparently doesn't quite play nicely with multiple inheritence for MixIn'\n    # of regexp based URLs\n    #@abstractmethod\n    #@classmethod\n    #def _str_to_fields(cls, ri_str):\n    #    raise NotImplementedError\n\n    #\n    # If any field is specified, URL is not considered 'False', i.e.\n    # non-existing, although may be we could/shout omit having only\n    # scheme or port specified since it doesn't point to any useful\n    # location\n    #\n\n    def __nonzero__(self):\n        fields = self._fields\n        return any(fields.values())\n\n    # for PY3\n    __bool__ = __nonzero__\n\n    #\n    # Helpers to deal with internal structures and conversions\n    #\n\n    def _set_from_fields(self, **fields):\n        unknown_fields = set(fields).difference(self._FIELDS)\n        if unknown_fields:\n            raise ValueError(\"Do not know about %s. Known fields for %s are: %s\"\n                             % (unknown_fields, self.__class__, self._FIELDS))\n\n        # encode dicts for query or fragment into\n        for f in {'query', 'fragment'}:\n            v = fields.get(f)\n            if isinstance(v, dict):\n\n                ev = urlencode(v)\n                # / is reserved char within query\n                if f == 'fragment' and '%2F' not in str(v):\n                    # but seems to be ok'ish within the fragment which is\n                    # the last element of URI and anyways used only by the\n                    # client (i.e. by us here if used to compose the URL)\n                    # so let's return / back for clarity if there were no\n                    # awkward %2F to startswith\n                    ev = ev.replace('%2F', '/')\n                fields[f] = ev\n\n        self._fields.update(fields)\n\n    #\n    # Quick comparators\n    #\n\n    def __eq__(self, other):\n        if not isinstance(other, RI):\n            other = RI(other)\n        return isinstance(other, self.__class__) and dict(other._fields) == dict(self._fields)\n\n    def __ne__(self, other):\n        return not (self == other)\n\n    def __getattribute__(self, item):\n        if item.startswith('_') or item not in self._FIELDS:\n            return super(RI, self).__getattribute__(item)\n        else:\n            return self._fields[item]\n\n\nclass URL(RI):\n    \"\"\"Universal resource locator\n\n    Although largely decorating urlparse.ParseResult, it\n    - doesn't mandate providing all parts of the URL\n    - doesn't require netloc but rather asks for separate username, password, and hostname\n    \"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'scheme',\n        'username',\n        'password',\n        'hostname', 'port',\n        'query',\n        'fragment',\n    )\n\n    def as_str(self):\n        \"\"\"Render URL as a string\"\"\"\n        return urlunparse(self.to_pr())\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        fields = URL._pr_to_fields(urlparse(url_str))\n        fields['path'] = urlunquote(fields['path'])\n        return fields\n\n    def to_pr(self):\n        \"\"\"Convert URL to urlparse.ParseResults namedtuple\"\"\"\n        return self._fields_to_pr(self._fields)\n\n    @classmethod\n    def _fields_to_pr(cls, fields):\n        \"\"\"Recompose back fields dict to ParseResult\"\"\"\n        netloc = fields['username'] or ''\n        if fields['password']:\n            netloc += ':' + fields['password']\n        if netloc:\n            netloc += '@'\n        netloc += fields['hostname']\n        if fields['port']:\n            if fields['hostname'].count(':') >= 2:\n                # ipv6 -- need to enclose in []\n                netloc = '[%s]:%s' % (netloc, fields['port'])\n            else:\n                netloc += ':%s' % fields['port']\n\n        pr_fields = {\n            f: fields[f]\n            for f in cls._FIELDS\n            if f not in ('hostname', 'password', 'username', 'port')\n        }\n        pr_fields['netloc'] = netloc\n        pr_fields['params'] = ''\n        # We need to quote the path\n        pr_fields['path'] = urlquote(pr_fields['path'])\n        # TODO: figure out what to do with query/fragment... one step at a time\n        return ParseResult(**pr_fields)\n\n    @classmethod\n    def _pr_to_fields(cls, pr):\n        \"\"\"ParseResult is a tuple so immutable, which complicates adjusting it\n\n        This function converts ParseResult into dict\"\"\"\n\n        if pr.params:\n            lgr.warning(\"ParseResults contains params %r, which will be ignored\"\n                        % (pr.params,))\n\n        hostname_port = pr.netloc.split('@')[-1]\n        is_ipv6 = hostname_port.count(':') >= 2\n        # can't use just pr._asdict since we care to ask those properties\n        # such as .port , .hostname etc\n        # Forcing '' instead of None since those properties (.hostname), .password,\n        # .username return None if not available and we decided to uniformize\n        if is_ipv6:\n            rem = re.match('\\[(?P<hostname>.*)\\]:(?P<port>\\d+)', hostname_port)\n            if rem:\n                hostname, port = rem.groups()\n                port = int(port)\n            else:\n                hostname, port = hostname_port, ''\n\n            def _getattr(pr, f):\n                \"\"\"Helper for custom handling in case of ipv6 addresses which blows\n                stock ParseResults logic\"\"\"\n                if f == 'port':\n                    # for now not supported at all, so\n                    return port\n                elif f == 'hostname':\n                    return hostname\n                else:\n                    return getattr(pr, f)\n        else:\n            _getattr = getattr\n\n        return {f: (_getattr(pr, f) or '') for f in cls._FIELDS}\n\n    #\n    # Access helpers\n    #\n\n    def _parse_qs(self, s, auto_delist=True):\n        \"\"\"Helper around parse_qs to strip unneeded 'list'ing etc and return a dict of key=values\"\"\"\n        if not s:\n            return {}\n        out = OrderedDict(parse_qsl(s, 1))\n        if not auto_delist:\n            return out\n        for k in out:\n            v = out[k]\n            if isinstance(v, list) and len(v) == 1:\n                v = v[0]\n                out[k] = None if v == '' else v\n        return out\n\n    @property\n    def query_dict(self):\n        return self._parse_qs(self.query)\n\n    @property\n    def fragment_dict(self):\n        return self._parse_qs(self.fragment)\n\n    @property\n    def localpath(self):\n        if self.scheme != 'file':\n            raise ValueError(\n                \"Non 'file://' URL cannot be resolved to a local path\")\n        hostname = self.hostname\n        if not (hostname in (None, '', 'localhost', '::1') \\\n                or hostname.startswith('127.')):\n            raise ValueError(\"file:// URL does not point to 'localhost'\")\n        return self.path\n\n\n\nclass PathRI(RI):\n    \"\"\"RI pointing to a (local) file/directory\"\"\"\n    def as_str(self):\n        return self.path\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        return dict(path=url_str)\n\n    @property\n    def localpath(self):\n        return self.path\n\n\nclass RegexBasedURLMixin(object):\n    \"\"\"Base class for URLs which we could simple parse using regular expressions\"\"\"\n\n    _REGEX = None\n\n    # not used ATM but possible ;)\n    # @classmethod\n    # def is_str_matches(cls, url_str):\n    #     return bool(cls._REGEX.match(url_str))\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        re_match = cls._REGEX.match(url_str)\n        if not re_match:\n            # TODO: custom error?\n            raise ValueError(\n                \"Possibly incorrectly determined string %r correspond to %s address\"\n                \" -- it failed matching regex. Dunno how to handle. Contact developers\"\n                % (cls, url_str,)\n            )\n        fields = cls._get_blank_fields()\n        fields.update({k: v for k, v in iteritems(re_match.groupdict()) if v})\n        cls._normalize_fields(fields)\n        return fields\n\n    @classmethod\n    def _normalize_fields(self, fields):\n        \"\"\"Helper to be ran if any of the fields need to be normalized after parsing\"\"\"\n        pass\n\n\nclass SSHRI(RI, RegexBasedURLMixin):\n    \"\"\"RI pointing to a remote location reachable via SSH\"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'username',\n        'hostname',\n        'port',\n    )\n\n    _REGEX = re.compile(r'((?P<username>\\S*)@)?(?P<hostname>[^:]+)(\\:(?P<path>.*))?$')\n\n    @classmethod\n    def _normalize_fields(cls, fields):\n        if fields['path'] and fields['path'].startswith('//'):\n            # Let's normalize for now to avoid multiple leading slashes\n            fields['path'] = '/' + fields['path'].lstrip('/')\n        # escape path so we have direct representation of the path to work with\n        fields['path'] = unescape_ssh_path(fields['path'])\n\n    def as_str(self):\n        fields = self.fields  # copy so we could escape symbols\n        url_fmt = '{hostname}'\n        if fields['username']:\n            url_fmt = \"{username}@\" + url_fmt\n        if fields['path']:\n            url_fmt += ':{path}'\n        fields['path'] = escape_ssh_path(fields['path'])\n        return url_fmt.format(**fields)\n\n    # TODO:\n    # we can \"support\" localhost:path as localpaths\n\n\nclass DataLadRI(RI, RegexBasedURLMixin):\n    \"\"\"RI pointing to datasets within central DataLad super-dataset\"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'remote',\n    )\n\n    # For now or forever we don't deal with any fragments or other special stuff\n    _REGEX = re.compile(r'//(?P<remote>[^\\s/]*)/(?P<path>.*)$')\n\n    # do they need to be normalized??? loosing track ...\n\n    def as_str(self):\n        return \"//{remote}/{path}\".format(**self._fields)\n\n    def as_git_url(self):\n        \"\"\"Dereference /// into original URLs which could be used by git for cloning\n\n        Returns\n        -------\n        str\n          URL string to reference the DataLadRI from its /// form\n        \"\"\"\n        if self.remote:\n            raise NotImplementedError(\"not supported ATM to reference additional remotes\")\n        return \"{}{}\".format(DATASETS_TOPURL, self.path)\n\n\ndef _split_colon(s, maxsplit=1):\n    \"\"\"Split on unescaped colon\"\"\"\n    return re.compile(r'(?<!\\\\):').split(s, maxsplit=maxsplit)\n\n# \\ should be first to deal with\n_SSH_ESCAPED_CHARACTERS = '\\\\#&;`|*?~<>^()[]{}$\\'\" '\n\n\n# TODO: RF using re.sub\ndef escape_ssh_path(path):\n    \"\"\"Escape all special characters present in the path\"\"\"\n    for c in _SSH_ESCAPED_CHARACTERS:\n        if c in path:\n            path = path.replace(c, '\\\\' + c)\n    return path\n\n\ndef unescape_ssh_path(path):\n    \"\"\"Un-escape all special characters present in the path\"\"\"\n    for c in _SSH_ESCAPED_CHARACTERS[::-1]:\n        if c in path:\n            path = path.replace('\\\\' + c, c)\n    return path\n\n\ndef parse_url_opts(url):\n    \"\"\"Given a string with url-style query, split into content before # and options as dict\"\"\"\n    url = URL(url)\n    # we need to filter out query and fragment to get the base url\n    fields = url.fields\n    fields.pop('query')\n    fields.pop('fragment')\n    opts = url.query_dict\n    return str(URL(**fields)), opts\n\n\n# TODO: should we just define URL.good_for_git or smth like that? ;)\n# although git also understands regular paths\ndef is_url(ri):\n    \"\"\"Returns whether argument is a resource identifier what datalad should treat as a URL\n\n    This includes ssh \"urls\" which git understands.\n\n    Parameters\n    ----------\n    ri : str or RI\n      The resource identifier (as a string or RI) to \"analyze\"\n    \"\"\"\n    if not isinstance(ri, RI):\n        try:\n            ri = RI(ri)\n        except:\n            return False\n    return isinstance(ri, (URL, SSHRI))\n\n\n# TODO: RF to remove duplication\ndef is_datalad_compat_ri(ri):\n    \"\"\"Returns whether argument is a resource identifier what datalad should treat as a URL\n\n    including its own DataLadRI\n    \"\"\"\n    if not isinstance(ri, RI):\n        try:\n            ri = RI(ri)\n        except:\n            return False\n    return isinstance(ri, (URL, SSHRI, DataLadRI))\n\n\n# TODO: better name? additionally may be move to SSHRI.is_valid() or sth.\ndef is_ssh(ri):\n    \"\"\"helper to determine, whether `ri` requires an SSH connection\n\n    Parameters\n    ----------\n    ri: str or RI\n\n    Returns\n    -------\n    bool\n    \"\"\"\n\n    # not exactly fitting the doc, but we actually can deal not necessarily with\n    # string or RI only, but with everything RI itself can deal with:\n    _ri = RI(ri) if not isinstance(ri, RI) else ri\n\n    return isinstance(_ri, SSHRI) or \\\n           (isinstance(_ri, URL) and _ri.scheme == 'ssh')\n\n\n#### windows workaround ###\n# TODO: There should be a better way\ndef get_local_file_url(fname):\n    \"\"\"Return OS specific URL pointing to a local file\n\n    Parameters\n    ----------\n    fname : string\n        Filename.  If not absolute, abspath is used\n    \"\"\"\n    fname = fname if isabs(fname) else abspath(fname)\n    if on_windows:\n        fname_rep = fname.replace('\\\\', '/')\n        furl = \"file:///%s\" % urlquote(fname_rep)\n        lgr.debug(\"Replaced '\\\\' in file\\'s url: %s\" % furl)\n    else:\n        # TODO:  need to fix for all the encoding etc\n        furl = str(URL(scheme='file', path=fname))\n    return furl",
            "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\nimport calendar\nimport email.utils\nimport gzip\nimport os\nimport re\nimport shutil\nimport time\nimport iso8601\n\nfrom collections import OrderedDict\nfrom os.path import abspath, isabs\n\nfrom six import string_types\nfrom six import iteritems\nfrom six.moves.urllib.request import urlopen, Request\nfrom six.moves.urllib.parse import quote as urlquote, unquote as urlunquote\nfrom six.moves.urllib.parse import urljoin, urlparse, urlsplit, urlunsplit, urlunparse, ParseResult\nfrom six.moves.urllib.parse import parse_qsl\nfrom six.moves.urllib.parse import urlencode\nfrom six.moves.urllib.error import URLError\n\nfrom datalad.utils import on_windows\nfrom datalad.consts import DATASETS_TOPURL\n\nfrom datalad.utils import auto_repr\n# TODO not sure what needs to use `six` here yet\nimport requests\n\nimport logging\nlgr = logging.getLogger('datalad.network')\n\n\ndef get_response_disposition_filename(s):\n    \"\"\"Given a string s as from HTTP Content-Disposition field in the response\n    return possibly present filename if any\n    \"\"\"\n    if not s:\n        return None\n    # If the response has Content-Disposition, try to get filename from it\n    cd = map(\n            lambda x: x.strip().split('=', 1) if '=' in x else [x.strip(), ''],\n            s.split(';')\n    )\n    # unify the key to be lower case and make it into a dict\n    cd = dict([[x[0].lower()] + x[1:] for x in cd])\n    if 'filename' in cd:\n        filename = cd['filename'].strip(\"\\\"'\")\n        return filename\n    return None\n\n\ndef get_url_disposition_filename(url, headers=None):\n    \"\"\"Get filename as possibly provided by the server in Content-Disposition\n    \"\"\"\n    if headers is None:\n        request = Request(url)\n        r = retry_urlopen(request)\n        # things are different in requests\n        if 'requests.' in str(r.__class__):\n            headers = r.headers\n        else:\n            headers = r.info()\n    else:\n        r = None\n    try:\n        return get_response_disposition_filename(headers.get('Content-Disposition', ''))\n    finally:\n        if r:\n            r.close()\n\n\ndef get_url_straight_filename(url, strip=[], allowdir=False):\n    \"\"\"Get file/dir name of the last path component of the URL\n\n    Parameters\n    ----------\n    strip: list, optional\n      If provided, listed names will not be considered and their\n      parent directory will be selected\n    allowdir: bool, optional\n      If url points to a \"directory\" (ends with /), empty string\n      would be returned unless allowdir is True, in which case the\n      name of the directory would be returned\n    \"\"\"\n    path = urlunquote(urlsplit(url).path)\n    path_parts = path.split('/')\n\n    if allowdir:\n        # strip empty ones\n        while len(path_parts) > 1 and not path_parts[-1]:\n            path_parts = path_parts[:-1]\n\n    if strip:\n        while path_parts and path_parts[-1] in strip:\n            path_parts = path_parts[:-1]\n\n    if path_parts:\n        return path_parts[-1]\n    else:\n        return None\n\n\ndef get_url_filename(url, headers=None, strip=[]):\n    \"\"\"Get filename from the url, first consulting server about Content-Disposition\n    \"\"\"\n    filename = get_url_disposition_filename(url, headers)\n    if filename:\n        return filename\n    return get_url_straight_filename(url, strip=[])\n\n\ndef get_url_response_stamp(url, response_info):\n    size, mtime = None, None\n    if 'Content-length' in response_info:\n        size = int(response_info['Content-length'])\n    if 'Last-modified' in response_info:\n        mtime = calendar.timegm(email.utils.parsedate(\n            response_info['Last-modified']))\n    return dict(size=size, mtime=mtime, url=url)\n\n\ndef get_tld(url):\n    \"\"\"Return top level domain from a url\n\n    Parameters\n    ----------\n    url : str\n    \"\"\"\n    # maybe use this instead to be safe:  https://pypi.python.org/pypi/tld\n    if not url.strip():\n        raise ValueError(\"Empty URL has no TLD\")\n    rec = urlsplit(url)\n    if not rec.netloc:\n        if not rec.scheme:\n            # There were no scheme provided thus netloc was empty -- must have been a simple 'path like'\n            return url.split('/', 1)[0]\n        else:\n            raise ValueError(\"It seems that only the scheme was provided without the net location/TLD\")\n    return rec.netloc\n\n\nfrom email.utils import parsedate_tz, mktime_tz\ndef rfc2822_to_epoch(datestr):\n    \"\"\"Given rfc2822 date/time format, return seconds since epoch\"\"\"\n    return mktime_tz(parsedate_tz(datestr))\n\n\nimport calendar\ndef iso8601_to_epoch(datestr):\n    \"\"\"Given ISO 8601 date/time format, return in seconds since epoch\n\n    iso8601 is used to parse properly the time zone information, which\n    can't be parsed with standard datetime strptime\n    \"\"\"\n    return calendar.timegm(iso8601.parse_date(datestr).timetuple())\n\n\ndef __urlopen_requests(url, header_vals=None):\n    # XXX Workaround for now for ... broken code\n    if isinstance(url, Request):\n        url = url.get_full_url()\n    return requests.Session().get(url)\n\n\ndef retry_urlopen(url, retries=3):\n    for t in range(retries):\n        try:\n            return __urlopen_requests(url)\n        except URLError as e:\n            lgr.warn(\"Received exception while reading %s: %s\" % (url, e))\n            if t == retries - 1:\n                # if we have reached allowed number of retries -- reraise\n                raise\n\n\ndef is_url_quoted(url):\n    \"\"\"Return either URL looks being already quoted\n    \"\"\"\n    try:\n        url_ = urlunquote(url)\n        return url != url_\n    except:  # problem with unquoting -- then it must be wasn't quoted (correctly)\n        return False\n\n\ndef same_website(url_rec, u_rec):\n    \"\"\"Decide either a link leads to external site\n\n    Parameters\n    ----------\n    url_rec: ParseResult\n      record for original url\n    u_rec: ParseResult\n      record for new url\n    \"\"\"\n    if isinstance(url_rec, string_types):\n        url_rec = urlparse(url_rec)\n    if isinstance(u_rec, string_types):\n        u_rec = urlparse(u_rec)\n    return (url_rec.netloc == u_rec.netloc)\n    # todo: collect more of sample cases.\n    # disabled below check while working on ratholeradio, since links\n    # could go to the parent and that is ok.  Figure out when it was\n    # desired not to go to the parent -- we might need explicit option\n    # and u_rec.path.startswith(url_rec.path)):\n\n\ndef dlurljoin(u_path, url):\n    url_rec = urlparse(url)  # probably duplicating parsing :-/ TODO\n    if url_rec.scheme:\n        # independent full url, so just return it\n        return url\n    if u_path.endswith('/'):  # should here be also a scheme use?\n        if url.startswith('/'):  # jump to the root\n            u_path_rec = urlparse(u_path)\n            return urljoin(urlunparse(\n                (u_path_rec.scheme, u_path_rec.netloc, '', '', '', '')), url)\n        else:\n            return os.path.join(u_path, url)\n    # TODO: recall where all this dirname came from and bring into the test\n    return urljoin(os.path.dirname(u_path) + '/', url)\n\n\n# TODO should it be a node maybe?\nclass SimpleURLStamper(object):\n    \"\"\"Gets a simple stamp about the URL: {url, time, size} of whatever was provided in the header\n    \"\"\"\n    def __init__(self, mode='full'):\n        self.mode = mode\n\n    def __call__(self, url):\n        # Extracted from above madness\n        # TODO: add mode alike to 'relaxed' where we would not\n        # care about content-disposition filename\n        # http://stackoverflow.com/questions/862173/how-to-download-a-file-using-python-in-a-smarter-way\n        request = Request(url)\n\n        # No traffic compression since we do not know how to identify\n        # exactly either it has to be decompressed\n        # request.add_header('Accept-encoding', 'gzip,deflate')\n        #\n        # TODO: think about stamping etc -- we seems to be redoing\n        # what git-annex does for us already... not really\n        r = retry_urlopen(request)\n        try:\n            r_info = r.info()\n            r_stamp = get_url_response_stamp(url, r_info)\n\n            return dict(mtime=r_stamp['mtime'], size=r_stamp['size'], url=url)\n        finally:\n            r.close()\n\n\n# TODO:  make it consistent/clear at what stage % encoding/decoding happens!\n# now it is a mix!\n\n#\n# Useful functionality in requests.models\n#  utils.requote_uri -- quote/unquote cycle to guarantee consistent appearance\n#  RequestEncodingMixin._encode_params -- Will successfully encode parameters when passed as a dict or a list of ...\n#  PreparedRequest().prepare_url(url, params) -- nicely cares about url encodings etc\n#\n\ndef _guess_ri_cls(ri):\n    \"\"\"Factory function which would determine which type of a ri a provided string is\"\"\"\n    # We assume that it is a URL and parse it. Depending on the result\n    # we might decide that it was something else ;)\n    fields = URL._pr_to_fields(urlparse(ri))\n    lgr.log(5, \"Parsed ri %s into fields %s\" % (ri, fields))\n\n    type_ = 'url'\n    # Special treatments\n    # file:///path should stay file:\n    if fields['scheme'] and fields['scheme'] not in {'file'} \\\n            and not fields['hostname']:\n        # dl+archive:... or just for ssh   hostname:path/p1\n        if '+' not in fields['scheme']:\n            type_ = 'ssh'\n            lgr.log(5, \"Assuming ssh style ri, adjusted: %s\" % (fields,))\n\n    if not fields['scheme'] and not fields['hostname']:\n        parts = _split_colon(ri)\n        if fields['path'] and '@' in fields['path'] or len(parts) > 1:\n            # user@host:path/sp1\n            # or host_name: (hence parts check)\n            # TODO: we need a regex to catch those really, parts check is not suff\n            type_ = 'ssh'\n        elif ri.startswith('//'):\n            # e.g. // or ///path\n            type_ = 'datalad'\n        else:\n            type_ = 'file'\n\n    if not fields['scheme'] and fields['hostname']:\n        # e.g. //a/path\n        type_ = 'datalad'\n\n    TYPES = {\n        'url': URL,\n        'ssh':  SSHRI,\n        'file': PathRI,\n        'datalad': DataLadRI\n    }\n    cls = TYPES[type_]\n    # just parse the ri according to regex matchint ssh \"ri\" specs\n    lgr.log(5, \"Detected %s ri\" % type_)\n    return cls\n\n\nclass RI(object):\n    \"\"\"Resource Identifier - base class and a factory for URL, SSHRI, etc\n\n    Intended to be a R/O object (i.e. no fields should be changed in-place).\n    Subclasses define specific collections of fields they care about in _FIELDS\n    class variable.\n    The idea is that this class should help to break apart a URL, while being\n    able to rebuild itself into a string representation for reuse\n\n    `RI` could be used as factory, whenever type of the resource is unknown and\n    must be guessed from the string representation.  One of the subclasses will be\n    provided as output, e.g.\n\n    >>> RI('http://example.com')\n    URL(hostname='example.com', scheme='http')\n    >>> RI('example.com:path')\n    SSHRI(hostname='example.com', path='path')\n    \"\"\"\n\n    # All of the subclasses will provide path\n    _FIELDS = (\n        'path',\n    )\n\n    __slots__ = _FIELDS + ('_fields', '_str')\n\n    def __new__(cls, ri=None, **kwargs):\n        \"\"\"Used as a possible factory for known RI types\n\n        Returns\n        -------\n        RI\n           uninitialized RI object of appropriate class with _str\n           set to string representation if was provided\n\n        \"\"\"\n        if cls is RI and ri is not None:\n            # RI class was used as a factory\n            cls = _guess_ri_cls(ri)\n\n        ri_obj = super(RI, cls).__new__(cls)\n        # Store internally original str\n        ri_obj._str = ri\n        return ri_obj\n\n    def __init__(self, ri=None, **fields):\n        \"\"\"\n        Parameters\n        ----------\n        ri: str, optional\n          String version of a resource specific for this class.  If you would like\n          a type of the resource be deduced, use RI(ri)\n        **fields: dict, optional\n          The values for the fields defined in _FIELDS class variable.\n        \"\"\"\n        if ri and (bool(ri) == bool(fields)):\n            raise ValueError(\n                \"Specify either ri or breakdown from the fields, not both. \"\n                \"Got ri=%r, fields=%r\" % (ri, fields))\n\n        self._fields = self._get_blank_fields()\n        if ri is not None:\n            fields = self._str_to_fields(ri)\n        self._set_from_fields(**fields)\n\n        # If was initialized from a string representation\n        if self._str is not None:\n            # well -- some ris might not unparse identically back\n            # strictly speaking, but let's assume they do\n            ri_ = self.as_str()\n            if ri != ri_:\n                lgr.warning(\"Parsed version of %s %r differs from original %r\",\n                            self.__class__.__name__, ri_, ri)\n\n    @classmethod\n    def _get_blank_fields(cls, **fields):\n        return OrderedDict(((f, fields.get(f, '')) for f in cls._FIELDS))\n\n    @property\n    def fields(self):\n        \"\"\"Returns shallow copy of fields to ease manipulations\"\"\"\n        return self._fields.copy()\n\n    def __repr__(self):\n        # since auto_repr doesn't support \"non-0\" values atm\n        return \"%s(%s)\" % (\n            self.__class__.__name__,\n            \", \".join([\"%s=%r\" % (k, v)\n                       for k, v in sorted(self._fields.items())\n                       if v]))\n\n    # Lazily evaluated if _str was not set\n    def __str__(self):\n        if self._str is None:\n            self._str = self.as_str()\n        return self._str\n\n    @classmethod\n    def from_str(cls, ri_str):\n        obj = cls(**cls._str_to_fields(ri_str))\n        obj._str = ri_str\n        return obj\n\n    @property\n    def localpath(self):\n        # by default RIs point to remote locations\n        raise ValueError(\"%s points to remote location\" % self)\n\n    # Apparently doesn't quite play nicely with multiple inheritence for MixIn'\n    # of regexp based URLs\n    #@abstractmethod\n    #@classmethod\n    #def _str_to_fields(cls, ri_str):\n    #    raise NotImplementedError\n\n    #\n    # If any field is specified, URL is not considered 'False', i.e.\n    # non-existing, although may be we could/shout omit having only\n    # scheme or port specified since it doesn't point to any useful\n    # location\n    #\n\n    def __nonzero__(self):\n        fields = self._fields\n        return any(fields.values())\n\n    # for PY3\n    __bool__ = __nonzero__\n\n    #\n    # Helpers to deal with internal structures and conversions\n    #\n\n    def _set_from_fields(self, **fields):\n        unknown_fields = set(fields).difference(self._FIELDS)\n        if unknown_fields:\n            raise ValueError(\"Do not know about %s. Known fields for %s are: %s\"\n                             % (unknown_fields, self.__class__, self._FIELDS))\n\n        # encode dicts for query or fragment into\n        for f in {'query', 'fragment'}:\n            v = fields.get(f)\n            if isinstance(v, dict):\n\n                ev = urlencode(v)\n                # / is reserved char within query\n                if f == 'fragment' and '%2F' not in str(v):\n                    # but seems to be ok'ish within the fragment which is\n                    # the last element of URI and anyways used only by the\n                    # client (i.e. by us here if used to compose the URL)\n                    # so let's return / back for clarity if there were no\n                    # awkward %2F to startswith\n                    ev = ev.replace('%2F', '/')\n                fields[f] = ev\n\n        self._fields.update(fields)\n\n    #\n    # Quick comparators\n    #\n\n    def __eq__(self, other):\n        if not isinstance(other, RI):\n            other = RI(other)\n        return isinstance(other, self.__class__) and dict(other._fields) == dict(self._fields)\n\n    def __ne__(self, other):\n        return not (self == other)\n\n    def __getattribute__(self, item):\n        if item.startswith('_') or item not in self._FIELDS:\n            return super(RI, self).__getattribute__(item)\n        else:\n            return self._fields[item]\n\n\nclass URL(RI):\n    \"\"\"Universal resource locator\n\n    Although largely decorating urlparse.ParseResult, it\n    - doesn't mandate providing all parts of the URL\n    - doesn't require netloc but rather asks for separate username, password, and hostname\n    \"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'scheme',\n        'username',\n        'password',\n        'hostname', 'port',\n        'query',\n        'fragment',\n    )\n\n    def as_str(self):\n        \"\"\"Render URL as a string\"\"\"\n        return urlunparse(self.to_pr())\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        fields = URL._pr_to_fields(urlparse(url_str))\n        fields['path'] = urlunquote(fields['path'])\n        return fields\n\n    def to_pr(self):\n        \"\"\"Convert URL to urlparse.ParseResults namedtuple\"\"\"\n        return self._fields_to_pr(self._fields)\n\n    @classmethod\n    def _fields_to_pr(cls, fields):\n        \"\"\"Recompose back fields dict to ParseResult\"\"\"\n        netloc = fields['username'] or ''\n        if fields['password']:\n            netloc += ':' + fields['password']\n        if netloc:\n            netloc += '@'\n        netloc += fields['hostname']\n        if fields['port']:\n            if fields['hostname'].count(':') >= 2:\n                # ipv6 -- need to enclose in []\n                netloc = '[%s]:%s' % (netloc, fields['port'])\n            else:\n                netloc += ':%s' % fields['port']\n\n        pr_fields = {\n            f: fields[f]\n            for f in cls._FIELDS\n            if f not in ('hostname', 'password', 'username', 'port')\n        }\n        pr_fields['netloc'] = netloc\n        pr_fields['params'] = ''\n        # We need to quote the path\n        pr_fields['path'] = urlquote(pr_fields['path'])\n        # TODO: figure out what to do with query/fragment... one step at a time\n        return ParseResult(**pr_fields)\n\n    @classmethod\n    def _pr_to_fields(cls, pr):\n        \"\"\"ParseResult is a tuple so immutable, which complicates adjusting it\n\n        This function converts ParseResult into dict\"\"\"\n\n        if pr.params:\n            lgr.warning(\"ParseResults contains params %r, which will be ignored\"\n                        % (pr.params,))\n\n        hostname_port = pr.netloc.split('@')[-1]\n        is_ipv6 = hostname_port.count(':') >= 2\n        # can't use just pr._asdict since we care to ask those properties\n        # such as .port , .hostname etc\n        # Forcing '' instead of None since those properties (.hostname), .password,\n        # .username return None if not available and we decided to uniformize\n        if is_ipv6:\n            rem = re.match('\\[(?P<hostname>.*)\\]:(?P<port>\\d+)', hostname_port)\n            if rem:\n                hostname, port = rem.groups()\n                port = int(port)\n            else:\n                hostname, port = hostname_port, ''\n\n            def _getattr(pr, f):\n                \"\"\"Helper for custom handling in case of ipv6 addresses which blows\n                stock ParseResults logic\"\"\"\n                if f == 'port':\n                    # for now not supported at all, so\n                    return port\n                elif f == 'hostname':\n                    return hostname\n                else:\n                    return getattr(pr, f)\n        else:\n            _getattr = getattr\n\n        return {f: (_getattr(pr, f) or '') for f in cls._FIELDS}\n\n    #\n    # Access helpers\n    #\n\n    def _parse_qs(self, s, auto_delist=True):\n        \"\"\"Helper around parse_qs to strip unneeded 'list'ing etc and return a dict of key=values\"\"\"\n        if not s:\n            return {}\n        out = OrderedDict(parse_qsl(s, 1))\n        if not auto_delist:\n            return out\n        for k in out:\n            v = out[k]\n            if isinstance(v, list) and len(v) == 1:\n                v = v[0]\n                out[k] = None if v == '' else v\n        return out\n\n    @property\n    def query_dict(self):\n        return self._parse_qs(self.query)\n\n    @property\n    def fragment_dict(self):\n        return self._parse_qs(self.fragment)\n\n    @property\n    def localpath(self):\n        if self.scheme != 'file':\n            raise ValueError(\n                \"Non 'file://' URL cannot be resolved to a local path\")\n        hostname = self.hostname\n        if not (hostname in (None, '', 'localhost', '::1') \\\n                or hostname.startswith('127.')):\n            raise ValueError(\"file:// URL does not point to 'localhost'\")\n        return self.path\n\n\n\nclass PathRI(RI):\n    \"\"\"RI pointing to a (local) file/directory\"\"\"\n    def as_str(self):\n        return self.path\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        return dict(path=url_str)\n\n    @property\n    def localpath(self):\n        return self.path\n\n\nclass RegexBasedURLMixin(object):\n    \"\"\"Base class for URLs which we could simple parse using regular expressions\"\"\"\n\n    _REGEX = None\n\n    # not used ATM but possible ;)\n    # @classmethod\n    # def is_str_matches(cls, url_str):\n    #     return bool(cls._REGEX.match(url_str))\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        re_match = cls._REGEX.match(url_str)\n        if not re_match:\n            # TODO: custom error?\n            raise ValueError(\n                \"Possibly incorrectly determined string %r correspond to %s address\"\n                \" -- it failed matching regex. Dunno how to handle. Contact developers\"\n                % (cls, url_str,)\n            )\n        fields = cls._get_blank_fields()\n        fields.update({k: v for k, v in iteritems(re_match.groupdict()) if v})\n        cls._normalize_fields(fields)\n        return fields\n\n    @classmethod\n    def _normalize_fields(self, fields):\n        \"\"\"Helper to be ran if any of the fields need to be normalized after parsing\"\"\"\n        pass\n\n\nclass SSHRI(RI, RegexBasedURLMixin):\n    \"\"\"RI pointing to a remote location reachable via SSH\"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'username',\n        'hostname',\n        'port',\n    )\n\n    _REGEX = re.compile(r'((?P<username>\\S*)@)?(?P<hostname>[^:]+)(\\:(?P<path>.*))?$')\n\n    @classmethod\n    def _normalize_fields(cls, fields):\n        if fields['path'] and fields['path'].startswith('//'):\n            # Let's normalize for now to avoid multiple leading slashes\n            fields['path'] = '/' + fields['path'].lstrip('/')\n        # escape path so we have direct representation of the path to work with\n        fields['path'] = unescape_ssh_path(fields['path'])\n\n    def as_str(self):\n        fields = self.fields  # copy so we could escape symbols\n        url_fmt = '{hostname}'\n        if fields['username']:\n            url_fmt = \"{username}@\" + url_fmt\n        if fields['path']:\n            url_fmt += ':{path}'\n        fields['path'] = escape_ssh_path(fields['path'])\n        return url_fmt.format(**fields)\n\n    # TODO:\n    # we can \"support\" localhost:path as localpaths\n\n\nclass DataLadRI(RI, RegexBasedURLMixin):\n    \"\"\"RI pointing to datasets within central DataLad super-dataset\"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'remote',\n    )\n\n    # For now or forever we don't deal with any fragments or other special stuff\n    _REGEX = re.compile(r'//(?P<remote>[^\\s/]*)/(?P<path>.*)$')\n\n    # do they need to be normalized??? loosing track ...\n\n    def as_str(self):\n        return \"//{remote}/{path}\".format(**self._fields)\n\n    def as_git_url(self):\n        \"\"\"Dereference /// into original URLs which could be used by git for cloning\n\n        Returns\n        -------\n        str\n          URL string to reference the DataLadRI from its /// form\n        \"\"\"\n        if self.remote:\n            raise NotImplementedError(\"not supported ATM to reference additional remotes\")\n        return \"{}{}\".format(DATASETS_TOPURL, self.path)\n\n\ndef _split_colon(s, maxsplit=1):\n    \"\"\"Split on unescaped colon\"\"\"\n    return re.compile(r'(?<!\\\\):').split(s, maxsplit=maxsplit)\n\n# \\ should be first to deal with\n_SSH_ESCAPED_CHARACTERS = '\\\\#&;`|*?~<>^()[]{}$\\'\" '\n\n\n# TODO: RF using re.sub\ndef escape_ssh_path(path):\n    \"\"\"Escape all special characters present in the path\"\"\"\n    for c in _SSH_ESCAPED_CHARACTERS:\n        if c in path:\n            path = path.replace(c, '\\\\' + c)\n    return path\n\n\ndef unescape_ssh_path(path):\n    \"\"\"Un-escape all special characters present in the path\"\"\"\n    for c in _SSH_ESCAPED_CHARACTERS[::-1]:\n        if c in path:\n            path = path.replace('\\\\' + c, c)\n    return path\n\n\ndef parse_url_opts(url):\n    \"\"\"Given a string with url-style query, split into content before # and options as dict\"\"\"\n    url = URL(url)\n    # we need to filter out query and fragment to get the base url\n    fields = url.fields\n    fields.pop('query')\n    fields.pop('fragment')\n    opts = url.query_dict\n    return str(URL(**fields)), opts\n\n\n# TODO: should we just define URL.good_for_git or smth like that? ;)\n# although git also understands regular paths\ndef is_url(ri):\n    \"\"\"Returns whether argument is a resource identifier what datalad should treat as a URL\n\n    This includes ssh \"urls\" which git understands.\n\n    Parameters\n    ----------\n    ri : str or RI\n      The resource identifier (as a string or RI) to \"analyze\"\n    \"\"\"\n    if not isinstance(ri, RI):\n        try:\n            ri = RI(ri)\n        except:\n            return False\n    return isinstance(ri, (URL, SSHRI))\n\n\n# TODO: RF to remove duplication\ndef is_datalad_compat_ri(ri):\n    \"\"\"Returns whether argument is a resource identifier what datalad should treat as a URL\n\n    including its own DataLadRI\n    \"\"\"\n    if not isinstance(ri, RI):\n        try:\n            ri = RI(ri)\n        except:\n            return False\n    return isinstance(ri, (URL, SSHRI, DataLadRI))\n\n\n# TODO: better name? additionally may be move to SSHRI.is_valid() or sth.\ndef is_ssh(ri):\n    \"\"\"helper to determine, whether `ri` requires an SSH connection\n\n    Parameters\n    ----------\n    ri: str or RI\n\n    Returns\n    -------\n    bool\n    \"\"\"\n\n    # not exactly fitting the doc, but we actually can deal not necessarily with\n    # string or RI only, but with everything RI itself can deal with:\n    _ri = RI(ri) if not isinstance(ri, RI) else ri\n\n    return isinstance(_ri, SSHRI) or \\\n           (isinstance(_ri, URL) and _ri.scheme == 'ssh')\n\n\n#### windows workaround ###\n# TODO: There should be a better way\ndef get_local_file_url(fname):\n    \"\"\"Return OS specific URL pointing to a local file\n\n    Parameters\n    ----------\n    fname : string\n        Filename.  If not absolute, abspath is used\n    \"\"\"\n    fname = fname if isabs(fname) else abspath(fname)\n    if on_windows:\n        fname_rep = fname.replace('\\\\', '/')\n        furl = \"file:///%s\" % urlquote(fname_rep)\n        lgr.debug(\"Replaced '\\\\' in file\\'s url: %s\" % furl)\n    else:\n        # TODO:  need to fix for all the encoding etc\n        furl = str(URL(scheme='file', path=fname))\n    return furl",
            "header": {
                "file": {
                    "path": "datalad/support/network.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "58da7f6ee",
                "new_commit_hash": "93e77667a",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/support/network.py"
            },
            "plus_file": {
                "path": "b/datalad/support/network.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 46,
                            "length": 9
                        },
                        "new_range": {
                            "start": 46,
                            "length": 12
                        },
                        "section": "def get_response_disposition_filename(s):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    if not s:",
                                "type": "context"
                            },
                            {
                                "content": "        return None",
                                "type": "context"
                            },
                            {
                                "content": "    # If the response has Content-Disposition, try to get filename from it",
                                "type": "context"
                            },
                            {
                                "content": "    cd = dict(map(",
                                "type": "deleted"
                            },
                            {
                                "content": "        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),",
                                "type": "deleted"
                            },
                            {
                                "content": "        s.split(';')))",
                                "type": "deleted"
                            },
                            {
                                "content": "    cd = map(",
                                "type": "added"
                            },
                            {
                                "content": "            lambda x: x.strip().split('=', 1) if '=' in x else [x.strip(), ''],",
                                "type": "added"
                            },
                            {
                                "content": "            s.split(';')",
                                "type": "added"
                            },
                            {
                                "content": "    )",
                                "type": "added"
                            },
                            {
                                "content": "    # unify the key to be lower case and make it into a dict",
                                "type": "added"
                            },
                            {
                                "content": "    cd = dict([[x[0].lower()] + x[1:] for x in cd])",
                                "type": "added"
                            },
                            {
                                "content": "    if 'filename' in cd:",
                                "type": "context"
                            },
                            {
                                "content": "        filename = cd['filename'].strip(\"\\\"'\")",
                                "type": "context"
                            },
                            {
                                "content": "        return filename",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/support/network.py",
                            "type": "function",
                            "name": "get_response_disposition_filename",
                            "content": "def get_response_disposition_filename(s):\n    \"\"\"Given a string s as from HTTP Content-Disposition field in the response\n    return possibly present filename if any\n    \"\"\"\n    if not s:\n        return None\n    # If the response has Content-Disposition, try to get filename from it\n    cd = dict(map(\n        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),\n        s.split(';')))\n    if 'filename' in cd:\n        filename = cd['filename'].strip(\"\\\"'\")\n        return filename\n    return None",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 42,
                            "end_lineno": 55,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "780607ffa876cd8fdfcb14c1292e9ded444132a0^",
    "new_commit_hash": "780607ffa876cd8fdfcb14c1292e9ded444132a0",
    "commit_message": "BF: make extraction of Content-Disposition be robust to casing",
    "commit_date": "2016-07-21T15:11:27-04:00",
    "metadata": {}
}
{
    "repo_name": "datalad",
    "new_commit_hash": "780607ffa876cd8fdfcb14c1292e9ded444132a0",
    "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\nfrom six import PY3\n\nimport os\nimport six.moves.builtins as __builtin__\nfrom os.path import join as opj\n\nfrom datalad.downloaders.tests.utils import get_test_providers\nfrom datalad.downloaders.base import DownloadError\nfrom datalad.downloaders.base import IncompleteDownloadError\nfrom datalad.downloaders.base import BaseDownloader\nfrom datalad.downloaders.credentials import UserPassword\nfrom datalad.downloaders.http import HTMLFormAuthenticator\nfrom datalad.downloaders.http import HTTPDownloader\nfrom datalad.support.network import get_url_straight_filename\nfrom datalad.tests.utils import with_fake_cookies_db\nfrom datalad.tests.utils import skip_if_no_network\nfrom datalad.tests.utils import with_testsui\n\n# BTW -- mock_open is not in mock on wheezy (Debian 7.x)\ntry:\n    if PY3:\n        raise ImportError(\"Not yet ready apparently: https://travis-ci.org/datalad/datalad/jobs/111659666\")\n    import httpretty\nexcept ImportError:\n    class NoHTTPPretty(object):\n       __bool__ = __nonzero__ = lambda s: False\n       activate = lambda s, t: t\n    httpretty = NoHTTPPretty()\n\nfrom mock import patch\nfrom datalad.tests.utils import SkipTest\nfrom datalad.tests.utils import assert_in\nfrom datalad.tests.utils import assert_not_in\nfrom datalad.tests.utils import assert_equal\nfrom datalad.tests.utils import assert_greater\nfrom datalad.tests.utils import assert_false\nfrom datalad.tests.utils import assert_raises\nfrom datalad.tests.utils import ok_file_has_content\nfrom datalad.tests.utils import serve_path_via_http, with_tree\nfrom datalad.tests.utils import swallow_logs\nfrom datalad.tests.utils import swallow_outputs\nfrom datalad.tests.utils import with_tempfile\nfrom datalad.tests.utils import use_cassette\nfrom datalad.tests.utils import skip_if\nfrom datalad.tests.utils import without_http_proxy\nfrom datalad.support.status import FileStatus\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 ;)\ndef fake_open(write_=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(*args, **kwargs):\n        return myfile\n    return myopen\n\n\ndef _raise_IOError(*args, **kwargs):\n    raise IOError(\"Testing here\")\n\n\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    # 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    # 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    with swallow_logs(), \\\n         patch.object(__builtin__, 'open', fake_open(write_=_raise_IOError)):\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_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@use_cassette('test_authenticate_external_portals', record_mode='once')\ndef test_authenticate_external_portals():\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    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\"\ntest_authenticate_external_portals.tags = ['external-portal', 'network']\n\n\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    yield check_download_external_url, \\\n          \"ftp://ftp.gnu.org/README\", \\\n          None, \\\n          \"This is ftp.gnu.org\"\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    assert_equal(\n        HTTPDownloader.get_status_from_headers({\n            'Content-Disposition': 'Attachment;Filename=\"Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip\"',\n        }).filename,\n        'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip')\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    _fixed_credentials = [\n        {'user': 'testlogin', 'password': 'testpassword'},\n        {'user': 'testlogin2', 'password': 'testpassword2'},\n        {'user': 'testlogin2', 'password': 'testpassword3'}]\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, '')\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@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_testsui(responses=['yes'])  # will request to reentry it\ndef test_HTMLFormAuthenticator_httpretty_authfail404(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, ['failed', '404'])\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_HTMLFormAuthenticator_httpretty_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, '')\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",
        "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...\n❌ Verification failed\nPython 3.9 installation failed, trying Python 3.7...\nVerifying installation...\nDatalad version: 0.2.3\n✅ Installation successful!\nSuccessfully installed Datalad using Python 3.7\n",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 25 packages in 153ms\nInstalled 25 packages in 4.37s\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\nerror: The build backend returned an error\n  Caused by: Call to `setuptools.build_meta:__legacy__.build_editable` failed (exit status: 1)\n\n[stderr]\nTraceback (most recent call last):\n  File \"<string>\", line 14, in <module>\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzU4IcX/lib/python3.9/site-packages/setuptools/build_meta.py\", line 483, in get_requires_for_build_editable\n    return self.get_requires_for_build_wheel(config_settings)\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzU4IcX/lib/python3.9/site-packages/setuptools/build_meta.py\", line 334, in get_requires_for_build_wheel\n    return self._get_build_requires(config_settings, requirements=[])\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzU4IcX/lib/python3.9/site-packages/setuptools/build_meta.py\", line 304, in _get_build_requires\n    self.run_setup()\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzU4IcX/lib/python3.9/site-packages/setuptools/build_meta.py\", line 522, in run_setup\n    super().run_setup(setup_script=setup_script)\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzU4IcX/lib/python3.9/site-packages/setuptools/build_meta.py\", line 320, in run_setup\n    exec(code, locals())\n  File \"<string>\", line 39, in <module>\nAttributeError: module 'platform' has no attribute 'dist'\n\nhint: This usually indicates a problem with the package or the build environment.\nerror: The build backend returned an error\n  Caused by: Call to `setuptools.build_meta:__legacy__.build_editable` failed (exit status: 1)\n\n[stderr]\nTraceback (most recent call last):\n  File \"<string>\", line 14, in <module>\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzqpD8u/lib/python3.9/site-packages/setuptools/build_meta.py\", line 483, in get_requires_for_build_editable\n    return self.get_requires_for_build_wheel(config_settings)\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzqpD8u/lib/python3.9/site-packages/setuptools/build_meta.py\", line 334, in get_requires_for_build_wheel\n    return self._get_build_requires(config_settings, requirements=[])\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzqpD8u/lib/python3.9/site-packages/setuptools/build_meta.py\", line 304, in _get_build_requires\n    self.run_setup()\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzqpD8u/lib/python3.9/site-packages/setuptools/build_meta.py\", line 522, in run_setup\n    super().run_setup(setup_script=setup_script)\n  File \"/home/gcpuser/.cache/uv/builds-v0/.tmpzqpD8u/lib/python3.9/site-packages/setuptools/build_meta.py\", line 320, in run_setup\n    exec(code, locals())\n  File \"<string>\", line 39, in <module>\nAttributeError: module 'platform' has no attribute 'dist'\n\nhint: This usually indicates a problem with the package or the build environment.\nTraceback (most recent call last):\n  File \"<string>\", line 1, in <module>\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/datalad/__init__.py\", line 20, in <module>\n    from .config import ConfigManager\n  File \"/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/datalad/config.py\", line 17, in <module>\n    from appdirs import AppDirs\nModuleNotFoundError: No module named 'appdirs'\nUsing CPython 3.7.9\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 26 packages in 247ms\nInstalled 26 packages in 3.88s\n + bids==0.0\n + coverage==7.2.7\n + docopt==0.6.2\n + exceptiongroup==1.2.2\n + fasteners==0.19\n + grabbit==0.2.6\n + importlib-metadata==6.7.0\n + iniconfig==2.0.0\n + nibabel==4.0.2\n + num2words==0.5.14\n + numpy==1.21.6\n + packaging==24.0\n + pandas==1.3.5\n + patsy==1.0.1\n + pluggy==1.2.0\n + pybids==0.6.5\n + pytest==7.4.4\n + pytest-cov==4.1.0\n + python-dateutil==2.9.0.post0\n + pytz==2024.2\n + scipy==1.7.3\n + setuptools==68.0.0\n + six==1.17.0\n + tomli==2.0.1\n + typing-extensions==4.7.1\n + zipp==3.15.0\nResolved 69 packages in 2.54s\nPrepared 1 package in 4.40s\nInstalled 63 packages in 434ms\n + appdirs==1.4.4\n + attrs==24.2.0\n + automat==22.10.0\n + beautifulsoup4==4.12.3\n + boto==2.49.0\n + certifi==2024.12.14\n + cffi==1.15.1\n + charset-normalizer==3.4.1\n + constantly==15.1.0\n + cryptography==44.0.0\n + cssselect==1.2.0\n + datalad==0.2.3 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0)\n + filelock==3.12.2\n + gitdb==4.0.12\n + gitpython==3.1.43\n + httpretty==1.1.4\n + humanize==4.6.0\n + hyperlink==21.0.0\n + idna==3.10\n + importlib-resources==5.12.0\n + incremental==22.10.0\n + iso8601==2.1.0\n + itemadapter==0.8.0\n + itemloaders==1.1.0\n + jaraco-classes==3.2.3\n + jeepney==0.8.0\n + jmespath==1.0.1\n + keyring==24.1.1\n + keyrings-alt==4.2.0\n + lxml==5.3.0\n + mock==5.1.0\n + more-itertools==9.1.0\n + msgpack-python==0.5.6\n + multidict==6.0.5\n + nose==1.3.7\n + parsel==1.8.1\n + patool==1.12\n + protego==0.3.0\n + pyasn1==0.5.1\n + pyasn1-modules==0.3.0\n + pycparser==2.21\n + pydispatcher==2.0.7\n + pyopenssl==24.3.0\n + pyyaml==6.0.1\n + queuelib==1.6.2\n + requests==2.31.0\n + requests-file==2.1.0\n + requests-ftp==0.3.1\n + scrapy==2.9.0\n + secretstorage==3.3.3\n + service-identity==21.1.0\n + smmap==5.0.2\n + soupsieve==2.4.1\n + testtools==2.7.1\n + tldextract==4.0.0\n + tqdm==4.67.1\n + twisted==23.8.0\n + urllib3==2.0.7\n + vcrpy==4.3.0\n + w3lib==2.1.2\n + wrapt==1.16.0\n + yarl==1.9.4\n + zope-interface==6.4.post2\nAudited 1 package in 77ms\n/home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/.venv/lib/python3.7/site-packages/urllib3/__init__.py:38: NotOpenSSLWarning: urllib3 v2.0 only supports OpenSSL 1.1.1+, currently the 'ssl' module is compiled with 'LibreSSL 3.1.4'. See: https://github.com/urllib3/urllib3/issues/3020\n  exceptions.NotOpenSSLWarning,\n",
    "new_commit_res_code": 1,
    "new_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.7.9, pytest-7.4.4, pluggy-1.2.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0\nplugins: cov-4.1.0\ncollected 9 items\n\nr2e_tests/test_1.py .E..E.EEE                                            [100%]\n\n==================================== ERRORS ====================================\n_________________ ERROR at setup of test_HTTPDownloader_basic __________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 86\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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:86\n_________________________ ERROR at setup of test_mtime _________________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 225\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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:225\n____________ ERROR at setup of test_HTMLFormAuthenticator_httpretty ____________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 287\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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:287\n______ ERROR at setup of test_HTMLFormAuthenticator_httpretty_authfail404 ______\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 354\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_testsui(responses=['yes'])  # will request to reentry it\n  def test_HTMLFormAuthenticator_httpretty_authfail404(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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:354\n___________ ERROR at setup of test_HTMLFormAuthenticator_httpretty_2 ___________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 401\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_HTMLFormAuthenticator_httpretty_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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:401\n==================================== PASSES ====================================\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_docstring\nPASSED r2e_tests/test_1.py::test_authenticate_external_portals\nPASSED r2e_tests/test_1.py::test_download_ftp\nPASSED r2e_tests/test_1.py::test_get_status_from_headers\nERROR r2e_tests/test_1.py::test_HTTPDownloader_basic\nERROR r2e_tests/test_1.py::test_mtime\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty_authfail404\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty_2\n========================= 4 passed, 5 errors in 1.15s ==========================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.7.9, pytest-7.4.4, pluggy-1.2.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0\nplugins: cov-4.1.0\ncollected 9 items\n\nr2e_tests/test_1.py .E..EFEEE                                            [100%]\n\n==================================== ERRORS ====================================\n_________________ ERROR at setup of test_HTTPDownloader_basic __________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 86\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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:86\n_________________________ ERROR at setup of test_mtime _________________________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 225\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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:225\n____________ ERROR at setup of test_HTMLFormAuthenticator_httpretty ____________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 287\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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:287\n______ ERROR at setup of test_HTMLFormAuthenticator_httpretty_authfail404 ______\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 354\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_testsui(responses=['yes'])  # will request to reentry it\n  def test_HTMLFormAuthenticator_httpretty_authfail404(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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:354\n___________ ERROR at setup of test_HTMLFormAuthenticator_httpretty_2 ___________\nfile /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py, line 401\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_HTMLFormAuthenticator_httpretty_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_780607ffa876cd8fdfcb14c1292e9ded444132a0/r2e_tests/test_1.py:401\n=================================== FAILURES ===================================\n_________________________ test_get_status_from_headers _________________________\n\n    def 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        assert_equal(\n            HTTPDownloader.get_status_from_headers({\n                'Content-Disposition': 'Attachment;Filename=\"Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip\"',\n            }).filename,\n>           'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip')\n\nr2e_tests/test_1.py:261: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n../../../../.local/share/uv/python/cpython-3.7.9-linux-x86_64-gnu/lib/python3.7/unittest/case.py:852: in assertEqual\n    assertion_func(first, second, msg=msg)\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n\nself = <nose.tools.trivial.Dummy testMethod=nop>, first = None\nsecond = 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'\nmsg = \"None != 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'\"\n\n    def _baseAssertEqual(self, first, second, msg=None):\n        \"\"\"The default assertEqual implementation, not type specific.\"\"\"\n        if not first == second:\n            standardMsg = '%s != %s' % _common_shorten_repr(first, second)\n            msg = self._formatMessage(msg, standardMsg)\n>           raise self.failureException(msg)\nE           AssertionError: None != 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'\n\n../../../../.local/share/uv/python/cpython-3.7.9-linux-x86_64-gnu/lib/python3.7/unittest/case.py:845: AssertionError\n==================================== PASSES ====================================\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_docstring\nPASSED r2e_tests/test_1.py::test_authenticate_external_portals\nPASSED r2e_tests/test_1.py::test_download_ftp\nERROR r2e_tests/test_1.py::test_HTTPDownloader_basic\nERROR r2e_tests/test_1.py::test_mtime\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty_authfail404\nERROR r2e_tests/test_1.py::test_HTMLFormAuthenticator_httpretty_2\nFAILED r2e_tests/test_1.py::test_get_status_from_headers - AssertionError: No...\n==================== 1 failed, 3 passed, 5 errors in 0.88s =====================\n",
    "old_commit_res_stderr": ""
}
datalad/downloaders/tests/test_http.py datalad/support/network.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title:** Filename extraction from Content-Disposition header fails with varied casing

**Description:**
When handling HTTP responses, extracting the filename from the `Content-Disposition` header fails if the casing of the parameters varies. For example, using "Filename" instead of "filename" results in the filename not being retrieved correctly.

**Example Code:**
```python
headers = {
    'Content-Disposition': 'attachment; Filename="Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip"'
}

filename = get_response_disposition_filename(headers.get('Content-Disposition'))
print(filename)  # Expected: 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'
```

**Expected Behavior:**
The function `get_response_disposition_filename` should correctly extract and return the filename regardless of the casing of the `Content-Disposition` parameters.

**Actual Behavior:**
When the `Content-Disposition` header contains parameters with different casing (e.g., "Filename" instead of "filename"), the function fails to extract the filename and returns `None`.

[/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/support/network.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.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##

import calendar
import email.utils
import gzip
import os
import re
import shutil
import time
import iso8601

from collections import OrderedDict
from os.path import abspath, isabs

from six import string_types
from six import iteritems
from six.moves.urllib.request import urlopen, Request
from six.moves.urllib.parse import quote as urlquote, unquote as urlunquote
from six.moves.urllib.parse import urljoin, urlparse, urlsplit, urlunsplit, urlunparse, ParseResult
from six.moves.urllib.parse import parse_qsl
from six.moves.urllib.parse import urlencode
from six.moves.urllib.error import URLError

from datalad.utils import on_windows
from datalad.consts import DATASETS_TOPURL

from datalad.utils import auto_repr
# TODO not sure what needs to use `six` here yet
import requests

import logging
lgr = logging.getLogger('datalad.network')


def get_response_disposition_filename(s):
    """Given a string s as from HTTP Content-Disposition field in the response
    return possibly present filename if any
    """
    if not s:
        return None
    # If the response has Content-Disposition, try to get filename from it
    cd = dict(map(
        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),
        s.split(';')))
    if 'filename' in cd:
        filename = cd['filename'].strip("\"'")
        return filename
    return None


def get_url_disposition_filename(url, headers=None):
    """Get filename as possibly provided by the server in Content-Disposition
    """
    if headers is None:
        request = Request(url)
        r = retry_urlopen(request)
        # things are different in requests
        if 'requests.' in str(r.__class__):
            headers = r.headers
        else:
            headers = r.info()
    else:
        r = None
    try:
        return get_response_disposition_filename(headers.get('Content-Disposition', ''))
    finally:
        if r:
            r.close()


def get_url_straight_filename(url, strip=[], allowdir=False):
    """Get file/dir name of the last path component of the URL

    Parameters
    ----------
    strip: list, optional
      If provided, listed names will not be considered and their
      parent directory will be selected
    allowdir: bool, optional
      If url points to a "directory" (ends with /), empty string
      would be returned unless allowdir is True, in which case the
      name of the directory would be returned
    """
    path = urlunquote(urlsplit(url).path)
    path_parts = path.split('/')

    if allowdir:
        # strip empty ones
        while len(path_parts) > 1 and not path_parts[-1]:
            path_parts = path_parts[:-1]

    if strip:
        while path_parts and path_parts[-1] in strip:
            path_parts = path_parts[:-1]

    if path_parts:
        return path_parts[-1]
    else:
        return None


def get_url_filename(url, headers=None, strip=[]):
    """Get filename from the url, first consulting server about Content-Disposition
    """
    filename = get_url_disposition_filename(url, headers)
    if filename:
        return filename
    return get_url_straight_filename(url, strip=[])


def get_url_response_stamp(url, response_info):
    size, mtime = None, None
    if 'Content-length' in response_info:
        size = int(response_info['Content-length'])
    if 'Last-modified' in response_info:
        mtime = calendar.timegm(email.utils.parsedate(
            response_info['Last-modified']))
    return dict(size=size, mtime=mtime, url=url)


def get_tld(url):
    """Return top level domain from a url

    Parameters
    ----------
    url : str
    """
    # maybe use this instead to be safe:  https://pypi.python.org/pypi/tld
    if not url.strip():
        raise ValueError("Empty URL has no TLD")
    rec = urlsplit(url)
    if not rec.netloc:
        if not rec.scheme:
            # There were no scheme provided thus netloc was empty -- must have been a simple 'path like'
            return url.split('/', 1)[0]
        else:
            raise ValueError("It seems that only the scheme was provided without the net location/TLD")
    return rec.netloc


from email.utils import parsedate_tz, mktime_tz
def rfc2822_to_epoch(datestr):
    """Given rfc2822 date/time format, return seconds since epoch"""
    return mktime_tz(parsedate_tz(datestr))


import calendar
def iso8601_to_epoch(datestr):
    """Given ISO 8601 date/time format, return in seconds since epoch

    iso8601 is used to parse properly the time zone information, which
    can't be parsed with standard datetime strptime
    """
    return calendar.timegm(iso8601.parse_date(datestr).timetuple())


def __urlopen_requests(url, header_vals=None):
    # XXX Workaround for now for ... broken code
    if isinstance(url, Request):
        url = url.get_full_url()
    return requests.Session().get(url)


def retry_urlopen(url, retries=3):
    for t in range(retries):
        try:
            return __urlopen_requests(url)
        except URLError as e:
            lgr.warn("Received exception while reading %s: %s" % (url, e))
            if t == retries - 1:
                # if we have reached allowed number of retries -- reraise
                raise


def is_url_quoted(url):
    """Return either URL looks being already quoted
    """
    try:
        url_ = urlunquote(url)
        return url != url_
    except:  # problem with unquoting -- then it must be wasn't quoted (correctly)
        return False


def same_website(url_rec, u_rec):
    """Decide either a link leads to external site

    Parameters
    ----------
    url_rec: ParseResult
      record for original url
    u_rec: ParseResult
      record for new url
    """
    if isinstance(url_rec, string_types):
        url_rec = urlparse(url_rec)
    if isinstance(u_rec, string_types):
        u_rec = urlparse(u_rec)
    return (url_rec.netloc == u_rec.netloc)
    # todo: collect more of sample cases.
    # disabled below check while working on ratholeradio, since links
    # could go to the parent and that is ok.  Figure out when it was
    # desired not to go to the parent -- we might need explicit option
    # and u_rec.path.startswith(url_rec.path)):


def dlurljoin(u_path, url):
    url_rec = urlparse(url)  # probably duplicating parsing :-/ TODO
    if url_rec.scheme:
        # independent full url, so just return it
        return url
    if u_path.endswith('/'):  # should here be also a scheme use?
        if url.startswith('/'):  # jump to the root
            u_path_rec = urlparse(u_path)
            return urljoin(urlunparse(
                (u_path_rec.scheme, u_path_rec.netloc, '', '', '', '')), url)
        else:
            return os.path.join(u_path, url)
    # TODO: recall where all this dirname came from and bring into the test
    return urljoin(os.path.dirname(u_path) + '/', url)


# TODO should it be a node maybe?
class SimpleURLStamper(object):
    """Gets a simple stamp about the URL: {url, time, size} of whatever was provided in the header
    """
    def __init__(self, mode='full'):
        self.mode = mode

    def __call__(self, url):
        # Extracted from above madness
        # TODO: add mode alike to 'relaxed' where we would not
        # care about content-disposition filename
        # http://stackoverflow.com/questions/862173/how-to-download-a-file-using-python-in-a-smarter-way
        request = Request(url)

        # No traffic compression since we do not know how to identify
        # exactly either it has to be decompressed
        # request.add_header('Accept-encoding', 'gzip,deflate')
        #
        # TODO: think about stamping etc -- we seems to be redoing
        # what git-annex does for us already... not really
        r = retry_urlopen(request)
        try:
            r_info = r.info()
            r_stamp = get_url_response_stamp(url, r_info)

            return dict(mtime=r_stamp['mtime'], size=r_stamp['size'], url=url)
        finally:
            r.close()


# TODO:  make it consistent/clear at what stage % encoding/decoding happens!
# now it is a mix!

#
# Useful functionality in requests.models
#  utils.requote_uri -- quote/unquote cycle to guarantee consistent appearance
#  RequestEncodingMixin._encode_params -- Will successfully encode parameters when passed as a dict or a list of ...
#  PreparedRequest().prepare_url(url, params) -- nicely cares about url encodings etc
#

def _guess_ri_cls(ri):
    """Factory function which would determine which type of a ri a provided string is"""
    # We assume that it is a URL and parse it. Depending on the result
    # we might decide that it was something else ;)
    fields = URL._pr_to_fields(urlparse(ri))
    lgr.log(5, "Parsed ri %s into fields %s" % (ri, fields))

    type_ = 'url'
    # Special treatments
    # file:///path should stay file:
    if fields['scheme'] and fields['scheme'] not in {'file'} \
            and not fields['hostname']:
        # dl+archive:... or just for ssh   hostname:path/p1
        if '+' not in fields['scheme']:
            type_ = 'ssh'
            lgr.log(5, "Assuming ssh style ri, adjusted: %s" % (fields,))

    if not fields['scheme'] and not fields['hostname']:
        parts = _split_colon(ri)
        if fields['path'] and '@' in fields['path'] or len(parts) > 1:
            # user@host:path/sp1
            # or host_name: (hence parts check)
            # TODO: we need a regex to catch those really, parts check is not suff
            type_ = 'ssh'
        elif ri.startswith('//'):
            # e.g. // or ///path
            type_ = 'datalad'
        else:
            type_ = 'file'

    if not fields['scheme'] and fields['hostname']:
        # e.g. //a/path
        type_ = 'datalad'

    TYPES = {
        'url': URL,
        'ssh':  SSHRI,
        'file': PathRI,
        'datalad': DataLadRI
    }
    cls = TYPES[type_]
    # just parse the ri according to regex matchint ssh "ri" specs
    lgr.log(5, "Detected %s ri" % type_)
    return cls


class RI(object):
    """Resource Identifier - base class and a factory for URL, SSHRI, etc

    Intended to be a R/O object (i.e. no fields should be changed in-place).
    Subclasses define specific collections of fields they care about in _FIELDS
    class variable.
    The idea is that this class should help to break apart a URL, while being
    able to rebuild itself into a string representation for reuse

    `RI` could be used as factory, whenever type of the resource is unknown and
    must be guessed from the string representation.  One of the subclasses will be
    provided as output, e.g.

    >>> RI('http://example.com')
    URL(hostname='example.com', scheme='http')
    >>> RI('example.com:path')
    SSHRI(hostname='example.com', path='path')
    """

    # All of the subclasses will provide path
    _FIELDS = (
        'path',
    )

    __slots__ = _FIELDS + ('_fields', '_str')

    def __new__(cls, ri=None, **kwargs):
        """Used as a possible factory for known RI types

        Returns
        -------
        RI
           uninitialized RI object of appropriate class with _str
           set to string representation if was provided

        """
        if cls is RI and ri is not None:
            # RI class was used as a factory
            cls = _guess_ri_cls(ri)

        ri_obj = super(RI, cls).__new__(cls)
        # Store internally original str
        ri_obj._str = ri
        return ri_obj

    def __init__(self, ri=None, **fields):
        """
        Parameters
        ----------
        ri: str, optional
          String version of a resource specific for this class.  If you would like
          a type of the resource be deduced, use RI(ri)
        **fields: dict, optional
          The values for the fields defined in _FIELDS class variable.
        """
        if ri and (bool(ri) == bool(fields)):
            raise ValueError(
                "Specify either ri or breakdown from the fields, not both. "
                "Got ri=%r, fields=%r" % (ri, fields))

        self._fields = self._get_blank_fields()
        if ri is not None:
            fields = self._str_to_fields(ri)
        self._set_from_fields(**fields)

        # If was initialized from a string representation
        if self._str is not None:
            # well -- some ris might not unparse identically back
            # strictly speaking, but let's assume they do
            ri_ = self.as_str()
            if ri != ri_:
                lgr.warning("Parsed version of %s %r differs from original %r",
                            self.__class__.__name__, ri_, ri)

    @classmethod
    def _get_blank_fields(cls, **fields):
        return OrderedDict(((f, fields.get(f, '')) for f in cls._FIELDS))

    @property
    def fields(self):
        """Returns shallow copy of fields to ease manipulations"""
        return self._fields.copy()

    def __repr__(self):
        # since auto_repr doesn't support "non-0" values atm
        return "%s(%s)" % (
            self.__class__.__name__,
            ", ".join(["%s=%r" % (k, v)
                       for k, v in sorted(self._fields.items())
                       if v]))

    # Lazily evaluated if _str was not set
    def __str__(self):
        if self._str is None:
            self._str = self.as_str()
        return self._str

    @classmethod
    def from_str(cls, ri_str):
        obj = cls(**cls._str_to_fields(ri_str))
        obj._str = ri_str
        return obj

    @property
    def localpath(self):
        # by default RIs point to remote locations
        raise ValueError("%s points to remote location" % self)

    # Apparently doesn't quite play nicely with multiple inheritence for MixIn'
    # of regexp based URLs
    #@abstractmethod
    #@classmethod
    #def _str_to_fields(cls, ri_str):
    #    raise NotImplementedError

    #
    # If any field is specified, URL is not considered 'False', i.e.
    # non-existing, although may be we could/shout omit having only
    # scheme or port specified since it doesn't point to any useful
    # location
    #

    def __nonzero__(self):
        fields = self._fields
        return any(fields.values())

    # for PY3
    __bool__ = __nonzero__

    #
    # Helpers to deal with internal structures and conversions
    #

    def _set_from_fields(self, **fields):
        unknown_fields = set(fields).difference(self._FIELDS)
        if unknown_fields:
            raise ValueError("Do not know about %s. Known fields for %s are: %s"
                             % (unknown_fields, self.__class__, self._FIELDS))

        # encode dicts for query or fragment into
        for f in {'query', 'fragment'}:
            v = fields.get(f)
            if isinstance(v, dict):

                ev = urlencode(v)
                # / is reserved char within query
                if f == 'fragment' and '%2F' not in str(v):
                    # but seems to be ok'ish within the fragment which is
                    # the last element of URI and anyways used only by the
                    # client (i.e. by us here if used to compose the URL)
                    # so let's return / back for clarity if there were no
                    # awkward %2F to startswith
                    ev = ev.replace('%2F', '/')
                fields[f] = ev

        self._fields.update(fields)

    #
    # Quick comparators
    #

    def __eq__(self, other):
        if not isinstance(other, RI):
            other = RI(other)
        return isinstance(other, self.__class__) and dict(other._fields) == dict(self._fields)

    def __ne__(self, other):
        return not (self == other)

    def __getattribute__(self, item):
        if item.startswith('_') or item not in self._FIELDS:
            return super(RI, self).__getattribute__(item)
        else:
            return self._fields[item]


class URL(RI):
    """Universal resource locator

    Although largely decorating urlparse.ParseResult, it
    - doesn't mandate providing all parts of the URL
    - doesn't require netloc but rather asks for separate username, password, and hostname
    """

    _FIELDS = RI._FIELDS + (
        'scheme',
        'username',
        'password',
        'hostname', 'port',
        'query',
        'fragment',
    )

    def as_str(self):
        """Render URL as a string"""
        return urlunparse(self.to_pr())

    @classmethod
    def _str_to_fields(cls, url_str):
        fields = URL._pr_to_fields(urlparse(url_str))
        fields['path'] = urlunquote(fields['path'])
        return fields

    def to_pr(self):
        """Convert URL to urlparse.ParseResults namedtuple"""
        return self._fields_to_pr(self._fields)

    @classmethod
    def _fields_to_pr(cls, fields):
        """Recompose back fields dict to ParseResult"""
        netloc = fields['username'] or ''
        if fields['password']:
            netloc += ':' + fields['password']
        if netloc:
            netloc += '@'
        netloc += fields['hostname']
        if fields['port']:
            if fields['hostname'].count(':') >= 2:
                # ipv6 -- need to enclose in []
                netloc = '[%s]:%s' % (netloc, fields['port'])
            else:
                netloc += ':%s' % fields['port']

        pr_fields = {
            f: fields[f]
            for f in cls._FIELDS
            if f not in ('hostname', 'password', 'username', 'port')
        }
        pr_fields['netloc'] = netloc
        pr_fields['params'] = ''
        # We need to quote the path
        pr_fields['path'] = urlquote(pr_fields['path'])
        # TODO: figure out what to do with query/fragment... one step at a time
        return ParseResult(**pr_fields)

    @classmethod
    def _pr_to_fields(cls, pr):
        """ParseResult is a tuple so immutable, which complicates adjusting it

        This function converts ParseResult into dict"""

        if pr.params:
            lgr.warning("ParseResults contains params %r, which will be ignored"
                        % (pr.params,))

        hostname_port = pr.netloc.split('@')[-1]
        is_ipv6 = hostname_port.count(':') >= 2
        # can't use just pr._asdict since we care to ask those properties
        # such as .port , .hostname etc
        # Forcing '' instead of None since those properties (.hostname), .password,
        # .username return None if not available and we decided to uniformize
        if is_ipv6:
            rem = re.match('\[(?P<hostname>.*)\]:(?P<port>\d+)', hostname_port)
            if rem:
                hostname, port = rem.groups()
                port = int(port)
            else:
                hostname, port = hostname_port, ''

            def _getattr(pr, f):
                """Helper for custom handling in case of ipv6 addresses which blows
                stock ParseResults logic"""
                if f == 'port':
                    # for now not supported at all, so
                    return port
                elif f == 'hostname':
                    return hostname
                else:
                    return getattr(pr, f)
        else:
            _getattr = getattr

        return {f: (_getattr(pr, f) or '') for f in cls._FIELDS}

    #
    # Access helpers
    #

    def _parse_qs(self, s, auto_delist=True):
        """Helper around parse_qs to strip unneeded 'list'ing etc and return a dict of key=values"""
        if not s:
            return {}
        out = OrderedDict(parse_qsl(s, 1))
        if not auto_delist:
            return out
        for k in out:
            v = out[k]
            if isinstance(v, list) and len(v) == 1:
                v = v[0]
                out[k] = None if v == '' else v
        return out

    @property
    def query_dict(self):
        return self._parse_qs(self.query)

    @property
    def fragment_dict(self):
        return self._parse_qs(self.fragment)

    @property
    def localpath(self):
        if self.scheme != 'file':
            raise ValueError(
                "Non 'file://' URL cannot be resolved to a local path")
        hostname = self.hostname
        if not (hostname in (None, '', 'localhost', '::1') \
                or hostname.startswith('127.')):
            raise ValueError("file:// URL does not point to 'localhost'")
        return self.path



class PathRI(RI):
    """RI pointing to a (local) file/directory"""
    def as_str(self):
        return self.path

    @classmethod
    def _str_to_fields(cls, url_str):
        return dict(path=url_str)

    @property
    def localpath(self):
        return self.path


class RegexBasedURLMixin(object):
    """Base class for URLs which we could simple parse using regular expressions"""

    _REGEX = None

    # not used ATM but possible ;)
    # @classmethod
    # def is_str_matches(cls, url_str):
    #     return bool(cls._REGEX.match(url_str))

    @classmethod
    def _str_to_fields(cls, url_str):
        re_match = cls._REGEX.match(url_str)
        if not re_match:
            # TODO: custom error?
            raise ValueError(
                "Possibly incorrectly determined string %r correspond to %s address"
                " -- it failed matching regex. Dunno how to handle. Contact developers"
                % (cls, url_str,)
            )
        fields = cls._get_blank_fields()
        fields.update({k: v for k, v in iteritems(re_match.groupdict()) if v})
        cls._normalize_fields(fields)
        return fields

    @classmethod
    def _normalize_fields(self, fields):
        """Helper to be ran if any of the fields need to be normalized after parsing"""
        pass


class SSHRI(RI, RegexBasedURLMixin):
    """RI pointing to a remote location reachable via SSH"""

    _FIELDS = RI._FIELDS + (
        'username',
        'hostname',
        'port',
    )

    _REGEX = re.compile(r'((?P<username>\S*)@)?(?P<hostname>[^:]+)(\:(?P<path>.*))?$')

    @classmethod
    def _normalize_fields(cls, fields):
        if fields['path'] and fields['path'].startswith('//'):
            # Let's normalize for now to avoid multiple leading slashes
            fields['path'] = '/' + fields['path'].lstrip('/')
        # escape path so we have direct representation of the path to work with
        fields['path'] = unescape_ssh_path(fields['path'])

    def as_str(self):
        fields = self.fields  # copy so we could escape symbols
        url_fmt = '{hostname}'
        if fields['username']:
            url_fmt = "{username}@" + url_fmt
        if fields['path']:
            url_fmt += ':{path}'
        fields['path'] = escape_ssh_path(fields['path'])
        return url_fmt.format(**fields)

    # TODO:
    # we can "support" localhost:path as localpaths


class DataLadRI(RI, RegexBasedURLMixin):
    """RI pointing to datasets within central DataLad super-dataset"""

    _FIELDS = RI._FIELDS + (
        'remote',
    )

    # For now or forever we don't deal with any fragments or other special stuff
    _REGEX = re.compile(r'//(?P<remote>[^\s/]*)/(?P<path>.*)$')

    # do they need to be normalized??? loosing track ...

    def as_str(self):
        return "//{remote}/{path}".format(**self._fields)

    def as_git_url(self):
        """Dereference /// into original URLs which could be used by git for cloning

        Returns
        -------
        str
          URL string to reference the DataLadRI from its /// form
        """
        if self.remote:
            raise NotImplementedError("not supported ATM to reference additional remotes")
        return "{}{}".format(DATASETS_TOPURL, self.path)


def _split_colon(s, maxsplit=1):
    """Split on unescaped colon"""
    return re.compile(r'(?<!\\):').split(s, maxsplit=maxsplit)

# \ should be first to deal with
_SSH_ESCAPED_CHARACTERS = '\\#&;`|*?~<>^()[]{}$\'" '


# TODO: RF using re.sub
def escape_ssh_path(path):
    """Escape all special characters present in the path"""
    for c in _SSH_ESCAPED_CHARACTERS:
        if c in path:
            path = path.replace(c, '\\' + c)
    return path


def unescape_ssh_path(path):
    """Un-escape all special characters present in the path"""
    for c in _SSH_ESCAPED_CHARACTERS[::-1]:
        if c in path:
            path = path.replace('\\' + c, c)
    return path


def parse_url_opts(url):
    """Given a string with url-style query, split into content before # and options as dict"""
    url = URL(url)
    # we need to filter out query and fragment to get the base url
    fields = url.fields
    fields.pop('query')
    fields.pop('fragment')
    opts = url.query_dict
    return str(URL(**fields)), opts


# TODO: should we just define URL.good_for_git or smth like that? ;)
# although git also understands regular paths
def is_url(ri):
    """Returns whether argument is a resource identifier what datalad should treat as a URL

    This includes ssh "urls" which git understands.

    Parameters
    ----------
    ri : str or RI
      The resource identifier (as a string or RI) to "analyze"
    """
    if not isinstance(ri, RI):
        try:
            ri = RI(ri)
        except:
            return False
    return isinstance(ri, (URL, SSHRI))


# TODO: RF to remove duplication
def is_datalad_compat_ri(ri):
    """Returns whether argument is a resource identifier what datalad should treat as a URL

    including its own DataLadRI
    """
    if not isinstance(ri, RI):
        try:
            ri = RI(ri)
        except:
            return False
    return isinstance(ri, (URL, SSHRI, DataLadRI))


# TODO: better name? additionally may be move to SSHRI.is_valid() or sth.
def is_ssh(ri):
    """helper to determine, whether `ri` requires an SSH connection

    Parameters
    ----------
    ri: str or RI

    Returns
    -------
    bool
    """

    # not exactly fitting the doc, but we actually can deal not necessarily with
    # string or RI only, but with everything RI itself can deal with:
    _ri = RI(ri) if not isinstance(ri, RI) else ri

    return isinstance(_ri, SSHRI) or \
           (isinstance(_ri, URL) and _ri.scheme == 'ssh')


#### windows workaround ###
# TODO: There should be a better way
def get_local_file_url(fname):
    """Return OS specific URL pointing to a local file

    Parameters
    ----------
    fname : string
        Filename.  If not absolute, abspath is used
    """
    fname = fname if isabs(fname) else abspath(fname)
    if on_windows:
        fname_rep = fname.replace('\\', '/')
        furl = "file:///%s" % urlquote(fname_rep)
        lgr.debug("Replaced '\\' in file\'s url: %s" % furl)
    else:
        # TODO:  need to fix for all the encoding etc
        furl = str(URL(scheme='file', path=fname))
    return furl
```
--- 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:** Filename extraction from Content-Disposition header fails with varied casing

**Description:**
When handling HTTP responses, extracting the filename from the `Content-Disposition` header fails if the casing of the parameters varies. For example, using "Filename" instead of "filename" results in the filename not being retrieved correctly.

**Example Code:**
```python
headers = {
    'Content-Disposition': 'attachment; Filename="Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip"'
}

filename = get_response_disposition_filename(headers.get('Content-Disposition'))
print(filename)  # Expected: 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'
```

**Expected Behavior:**
The function `get_response_disposition_filename` should correctly extract and return the filename regardless of the casing of the `Content-Disposition` parameters.

**Actual Behavior:**
When the `Content-Disposition` header contains parameters with different casing (e.g., "Filename" instead of "filename"), the function fails to extract the filename and returns `None`.

[/ISSUE]
{
    "test_docstring": "PASSED",
    "test_download_ftp": "PASSED",
    "test_get_status_from_headers": "PASSED",
    "test_HTTPDownloader_basic": "ERROR",
    "test_mtime": "ERROR",
    "test_HTMLFormAuthenticator_httpretty": "ERROR",
    "test_HTMLFormAuthenticator_httpretty_authfail404": "ERROR",
    "test_HTMLFormAuthenticator_httpretty_2": "ERROR"
}
diff --git a/datalad/support/network.py b/datalad/support/network.py
index 58da7f6ee..93e77667a 100644
--- a/datalad/support/network.py
+++ b/datalad/support/network.py
@@ -46,9 +46,12 @@ def get_response_disposition_filename(s):
     if not s:
         return None
     # If the response has Content-Disposition, try to get filename from it
-    cd = dict(map(
-        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),
-        s.split(';')))
+    cd = map(
+            lambda x: x.strip().split('=', 1) if '=' in x else [x.strip(), ''],
+            s.split(';')
+    )
+    # unify the key to be lower case and make it into a dict
+    cd = dict([[x[0].lower()] + x[1:] for x in cd])
     if 'filename' in cd:
         filename = cd['filename'].strip("\"'")
         return filename
 

{"datalad/support/network.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\nimport calendar\nimport email.utils\nimport gzip\nimport os\nimport re\nimport shutil\nimport time\nimport iso8601\n\nfrom collections import OrderedDict\nfrom os.path import abspath, isabs\n\nfrom six import string_types\nfrom six import iteritems\nfrom six.moves.urllib.request import urlopen, Request\nfrom six.moves.urllib.parse import quote as urlquote, unquote as urlunquote\nfrom six.moves.urllib.parse import urljoin, urlparse, urlsplit, urlunsplit, urlunparse, ParseResult\nfrom six.moves.urllib.parse import parse_qsl\nfrom six.moves.urllib.parse import urlencode\nfrom six.moves.urllib.error import URLError\n\nfrom datalad.utils import on_windows\nfrom datalad.consts import DATASETS_TOPURL\n\nfrom datalad.utils import auto_repr\n# TODO not sure what needs to use `six` here yet\nimport requests\n\nimport logging\nlgr = logging.getLogger('datalad.network')\n\n\ndef get_response_disposition_filename(s):\n    \"\"\"Given a string s as from HTTP Content-Disposition field in the response\n    return possibly present filename if any\n    \"\"\"\n    if not s:\n        return None\n    # If the response has Content-Disposition, try to get filename from it\n    cd = dict(map(\n        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),\n        s.split(';')))\n    if 'filename' in cd:\n        filename = cd['filename'].strip(\"\\\"'\")\n        return filename\n    return None\n\n\ndef get_url_disposition_filename(url, headers=None):\n    \"\"\"Get filename as possibly provided by the server in Content-Disposition\n    \"\"\"\n    if headers is None:\n        request = Request(url)\n        r = retry_urlopen(request)\n        # things are different in requests\n        if 'requests.' in str(r.__class__):\n            headers = r.headers\n        else:\n            headers = r.info()\n    else:\n        r = None\n    try:\n        return get_response_disposition_filename(headers.get('Content-Disposition', ''))\n    finally:\n        if r:\n            r.close()\n\n\ndef get_url_straight_filename(url, strip=[], allowdir=False):\n    \"\"\"Get file/dir name of the last path component of the URL\n\n    Parameters\n    ----------\n    strip: list, optional\n      If provided, listed names will not be considered and their\n      parent directory will be selected\n    allowdir: bool, optional\n      If url points to a \"directory\" (ends with /), empty string\n      would be returned unless allowdir is True, in which case the\n      name of the directory would be returned\n    \"\"\"\n    path = urlunquote(urlsplit(url).path)\n    path_parts = path.split('/')\n\n    if allowdir:\n        # strip empty ones\n        while len(path_parts) > 1 and not path_parts[-1]:\n            path_parts = path_parts[:-1]\n\n    if strip:\n        while path_parts and path_parts[-1] in strip:\n            path_parts = path_parts[:-1]\n\n    if path_parts:\n        return path_parts[-1]\n    else:\n        return None\n\n\ndef get_url_filename(url, headers=None, strip=[]):\n    \"\"\"Get filename from the url, first consulting server about Content-Disposition\n    \"\"\"\n    filename = get_url_disposition_filename(url, headers)\n    if filename:\n        return filename\n    return get_url_straight_filename(url, strip=[])\n\n\ndef get_url_response_stamp(url, response_info):\n    size, mtime = None, None\n    if 'Content-length' in response_info:\n        size = int(response_info['Content-length'])\n    if 'Last-modified' in response_info:\n        mtime = calendar.timegm(email.utils.parsedate(\n            response_info['Last-modified']))\n    return dict(size=size, mtime=mtime, url=url)\n\n\ndef get_tld(url):\n    \"\"\"Return top level domain from a url\n\n    Parameters\n    ----------\n    url : str\n    \"\"\"\n    # maybe use this instead to be safe:  https://pypi.python.org/pypi/tld\n    if not url.strip():\n        raise ValueError(\"Empty URL has no TLD\")\n    rec = urlsplit(url)\n    if not rec.netloc:\n        if not rec.scheme:\n            # There were no scheme provided thus netloc was empty -- must have been a simple 'path like'\n            return url.split('/', 1)[0]\n        else:\n            raise ValueError(\"It seems that only the scheme was provided without the net location/TLD\")\n    return rec.netloc\n\n\nfrom email.utils import parsedate_tz, mktime_tz\ndef rfc2822_to_epoch(datestr):\n    \"\"\"Given rfc2822 date/time format, return seconds since epoch\"\"\"\n    return mktime_tz(parsedate_tz(datestr))\n\n\nimport calendar\ndef iso8601_to_epoch(datestr):\n    \"\"\"Given ISO 8601 date/time format, return in seconds since epoch\n\n    iso8601 is used to parse properly the time zone information, which\n    can't be parsed with standard datetime strptime\n    \"\"\"\n    return calendar.timegm(iso8601.parse_date(datestr).timetuple())\n\n\ndef __urlopen_requests(url, header_vals=None):\n    # XXX Workaround for now for ... broken code\n    if isinstance(url, Request):\n        url = url.get_full_url()\n    return requests.Session().get(url)\n\n\ndef retry_urlopen(url, retries=3):\n    for t in range(retries):\n        try:\n            return __urlopen_requests(url)\n        except URLError as e:\n            lgr.warn(\"Received exception while reading %s: %s\" % (url, e))\n            if t == retries - 1:\n                # if we have reached allowed number of retries -- reraise\n                raise\n\n\ndef is_url_quoted(url):\n    \"\"\"Return either URL looks being already quoted\n    \"\"\"\n    try:\n        url_ = urlunquote(url)\n        return url != url_\n    except:  # problem with unquoting -- then it must be wasn't quoted (correctly)\n        return False\n\n\ndef same_website(url_rec, u_rec):\n    \"\"\"Decide either a link leads to external site\n\n    Parameters\n    ----------\n    url_rec: ParseResult\n      record for original url\n    u_rec: ParseResult\n      record for new url\n    \"\"\"\n    if isinstance(url_rec, string_types):\n        url_rec = urlparse(url_rec)\n    if isinstance(u_rec, string_types):\n        u_rec = urlparse(u_rec)\n    return (url_rec.netloc == u_rec.netloc)\n    # todo: collect more of sample cases.\n    # disabled below check while working on ratholeradio, since links\n    # could go to the parent and that is ok.  Figure out when it was\n    # desired not to go to the parent -- we might need explicit option\n    # and u_rec.path.startswith(url_rec.path)):\n\n\ndef dlurljoin(u_path, url):\n    url_rec = urlparse(url)  # probably duplicating parsing :-/ TODO\n    if url_rec.scheme:\n        # independent full url, so just return it\n        return url\n    if u_path.endswith('/'):  # should here be also a scheme use?\n        if url.startswith('/'):  # jump to the root\n            u_path_rec = urlparse(u_path)\n            return urljoin(urlunparse(\n                (u_path_rec.scheme, u_path_rec.netloc, '', '', '', '')), url)\n        else:\n            return os.path.join(u_path, url)\n    # TODO: recall where all this dirname came from and bring into the test\n    return urljoin(os.path.dirname(u_path) + '/', url)\n\n\n# TODO should it be a node maybe?\nclass SimpleURLStamper(object):\n    \"\"\"Gets a simple stamp about the URL: {url, time, size} of whatever was provided in the header\n    \"\"\"\n    def __init__(self, mode='full'):\n        self.mode = mode\n\n    def __call__(self, url):\n        # Extracted from above madness\n        # TODO: add mode alike to 'relaxed' where we would not\n        # care about content-disposition filename\n        # http://stackoverflow.com/questions/862173/how-to-download-a-file-using-python-in-a-smarter-way\n        request = Request(url)\n\n        # No traffic compression since we do not know how to identify\n        # exactly either it has to be decompressed\n        # request.add_header('Accept-encoding', 'gzip,deflate')\n        #\n        # TODO: think about stamping etc -- we seems to be redoing\n        # what git-annex does for us already... not really\n        r = retry_urlopen(request)\n        try:\n            r_info = r.info()\n            r_stamp = get_url_response_stamp(url, r_info)\n\n            return dict(mtime=r_stamp['mtime'], size=r_stamp['size'], url=url)\n        finally:\n            r.close()\n\n\n# TODO:  make it consistent/clear at what stage % encoding/decoding happens!\n# now it is a mix!\n\n#\n# Useful functionality in requests.models\n#  utils.requote_uri -- quote/unquote cycle to guarantee consistent appearance\n#  RequestEncodingMixin._encode_params -- Will successfully encode parameters when passed as a dict or a list of ...\n#  PreparedRequest().prepare_url(url, params) -- nicely cares about url encodings etc\n#\n\ndef _guess_ri_cls(ri):\n    \"\"\"Factory function which would determine which type of a ri a provided string is\"\"\"\n    # We assume that it is a URL and parse it. Depending on the result\n    # we might decide that it was something else ;)\n    fields = URL._pr_to_fields(urlparse(ri))\n    lgr.log(5, \"Parsed ri %s into fields %s\" % (ri, fields))\n\n    type_ = 'url'\n    # Special treatments\n    # file:///path should stay file:\n    if fields['scheme'] and fields['scheme'] not in {'file'} \\\n            and not fields['hostname']:\n        # dl+archive:... or just for ssh   hostname:path/p1\n        if '+' not in fields['scheme']:\n            type_ = 'ssh'\n            lgr.log(5, \"Assuming ssh style ri, adjusted: %s\" % (fields,))\n\n    if not fields['scheme'] and not fields['hostname']:\n        parts = _split_colon(ri)\n        if fields['path'] and '@' in fields['path'] or len(parts) > 1:\n            # user@host:path/sp1\n            # or host_name: (hence parts check)\n            # TODO: we need a regex to catch those really, parts check is not suff\n            type_ = 'ssh'\n        elif ri.startswith('//'):\n            # e.g. // or ///path\n            type_ = 'datalad'\n        else:\n            type_ = 'file'\n\n    if not fields['scheme'] and fields['hostname']:\n        # e.g. //a/path\n        type_ = 'datalad'\n\n    TYPES = {\n        'url': URL,\n        'ssh':  SSHRI,\n        'file': PathRI,\n        'datalad': DataLadRI\n    }\n    cls = TYPES[type_]\n    # just parse the ri according to regex matchint ssh \"ri\" specs\n    lgr.log(5, \"Detected %s ri\" % type_)\n    return cls\n\n\nclass RI(object):\n    \"\"\"Resource Identifier - base class and a factory for URL, SSHRI, etc\n\n    Intended to be a R/O object (i.e. no fields should be changed in-place).\n    Subclasses define specific collections of fields they care about in _FIELDS\n    class variable.\n    The idea is that this class should help to break apart a URL, while being\n    able to rebuild itself into a string representation for reuse\n\n    `RI` could be used as factory, whenever type of the resource is unknown and\n    must be guessed from the string representation.  One of the subclasses will be\n    provided as output, e.g.\n\n    >>> RI('http://example.com')\n    URL(hostname='example.com', scheme='http')\n    >>> RI('example.com:path')\n    SSHRI(hostname='example.com', path='path')\n    \"\"\"\n\n    # All of the subclasses will provide path\n    _FIELDS = (\n        'path',\n    )\n\n    __slots__ = _FIELDS + ('_fields', '_str')\n\n    def __new__(cls, ri=None, **kwargs):\n        \"\"\"Used as a possible factory for known RI types\n\n        Returns\n        -------\n        RI\n           uninitialized RI object of appropriate class with _str\n           set to string representation if was provided\n\n        \"\"\"\n        if cls is RI and ri is not None:\n            # RI class was used as a factory\n            cls = _guess_ri_cls(ri)\n\n        ri_obj = super(RI, cls).__new__(cls)\n        # Store internally original str\n        ri_obj._str = ri\n        return ri_obj\n\n    def __init__(self, ri=None, **fields):\n        \"\"\"\n        Parameters\n        ----------\n        ri: str, optional\n          String version of a resource specific for this class.  If you would like\n          a type of the resource be deduced, use RI(ri)\n        **fields: dict, optional\n          The values for the fields defined in _FIELDS class variable.\n        \"\"\"\n        if ri and (bool(ri) == bool(fields)):\n            raise ValueError(\n                \"Specify either ri or breakdown from the fields, not both. \"\n                \"Got ri=%r, fields=%r\" % (ri, fields))\n\n        self._fields = self._get_blank_fields()\n        if ri is not None:\n            fields = self._str_to_fields(ri)\n        self._set_from_fields(**fields)\n\n        # If was initialized from a string representation\n        if self._str is not None:\n            # well -- some ris might not unparse identically back\n            # strictly speaking, but let's assume they do\n            ri_ = self.as_str()\n            if ri != ri_:\n                lgr.warning(\"Parsed version of %s %r differs from original %r\",\n                            self.__class__.__name__, ri_, ri)\n\n    @classmethod\n    def _get_blank_fields(cls, **fields):\n        return OrderedDict(((f, fields.get(f, '')) for f in cls._FIELDS))\n\n    @property\n    def fields(self):\n        \"\"\"Returns shallow copy of fields to ease manipulations\"\"\"\n        return self._fields.copy()\n\n    def __repr__(self):\n        # since auto_repr doesn't support \"non-0\" values atm\n        return \"%s(%s)\" % (\n            self.__class__.__name__,\n            \", \".join([\"%s=%r\" % (k, v)\n                       for k, v in sorted(self._fields.items())\n                       if v]))\n\n    # Lazily evaluated if _str was not set\n    def __str__(self):\n        if self._str is None:\n            self._str = self.as_str()\n        return self._str\n\n    @classmethod\n    def from_str(cls, ri_str):\n        obj = cls(**cls._str_to_fields(ri_str))\n        obj._str = ri_str\n        return obj\n\n    @property\n    def localpath(self):\n        # by default RIs point to remote locations\n        raise ValueError(\"%s points to remote location\" % self)\n\n    # Apparently doesn't quite play nicely with multiple inheritence for MixIn'\n    # of regexp based URLs\n    #@abstractmethod\n    #@classmethod\n    #def _str_to_fields(cls, ri_str):\n    #    raise NotImplementedError\n\n    #\n    # If any field is specified, URL is not considered 'False', i.e.\n    # non-existing, although may be we could/shout omit having only\n    # scheme or port specified since it doesn't point to any useful\n    # location\n    #\n\n    def __nonzero__(self):\n        fields = self._fields\n        return any(fields.values())\n\n    # for PY3\n    __bool__ = __nonzero__\n\n    #\n    # Helpers to deal with internal structures and conversions\n    #\n\n    def _set_from_fields(self, **fields):\n        unknown_fields = set(fields).difference(self._FIELDS)\n        if unknown_fields:\n            raise ValueError(\"Do not know about %s. Known fields for %s are: %s\"\n                             % (unknown_fields, self.__class__, self._FIELDS))\n\n        # encode dicts for query or fragment into\n        for f in {'query', 'fragment'}:\n            v = fields.get(f)\n            if isinstance(v, dict):\n\n                ev = urlencode(v)\n                # / is reserved char within query\n                if f == 'fragment' and '%2F' not in str(v):\n                    # but seems to be ok'ish within the fragment which is\n                    # the last element of URI and anyways used only by the\n                    # client (i.e. by us here if used to compose the URL)\n                    # so let's return / back for clarity if there were no\n                    # awkward %2F to startswith\n                    ev = ev.replace('%2F', '/')\n                fields[f] = ev\n\n        self._fields.update(fields)\n\n    #\n    # Quick comparators\n    #\n\n    def __eq__(self, other):\n        if not isinstance(other, RI):\n            other = RI(other)\n        return isinstance(other, self.__class__) and dict(other._fields) == dict(self._fields)\n\n    def __ne__(self, other):\n        return not (self == other)\n\n    def __getattribute__(self, item):\n        if item.startswith('_') or item not in self._FIELDS:\n            return super(RI, self).__getattribute__(item)\n        else:\n            return self._fields[item]\n\n\nclass URL(RI):\n    \"\"\"Universal resource locator\n\n    Although largely decorating urlparse.ParseResult, it\n    - doesn't mandate providing all parts of the URL\n    - doesn't require netloc but rather asks for separate username, password, and hostname\n    \"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'scheme',\n        'username',\n        'password',\n        'hostname', 'port',\n        'query',\n        'fragment',\n    )\n\n    def as_str(self):\n        \"\"\"Render URL as a string\"\"\"\n        return urlunparse(self.to_pr())\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        fields = URL._pr_to_fields(urlparse(url_str))\n        fields['path'] = urlunquote(fields['path'])\n        return fields\n\n    def to_pr(self):\n        \"\"\"Convert URL to urlparse.ParseResults namedtuple\"\"\"\n        return self._fields_to_pr(self._fields)\n\n    @classmethod\n    def _fields_to_pr(cls, fields):\n        \"\"\"Recompose back fields dict to ParseResult\"\"\"\n        netloc = fields['username'] or ''\n        if fields['password']:\n            netloc += ':' + fields['password']\n        if netloc:\n            netloc += '@'\n        netloc += fields['hostname']\n        if fields['port']:\n            if fields['hostname'].count(':') >= 2:\n                # ipv6 -- need to enclose in []\n                netloc = '[%s]:%s' % (netloc, fields['port'])\n            else:\n                netloc += ':%s' % fields['port']\n\n        pr_fields = {\n            f: fields[f]\n            for f in cls._FIELDS\n            if f not in ('hostname', 'password', 'username', 'port')\n        }\n        pr_fields['netloc'] = netloc\n        pr_fields['params'] = ''\n        # We need to quote the path\n        pr_fields['path'] = urlquote(pr_fields['path'])\n        # TODO: figure out what to do with query/fragment... one step at a time\n        return ParseResult(**pr_fields)\n\n    @classmethod\n    def _pr_to_fields(cls, pr):\n        \"\"\"ParseResult is a tuple so immutable, which complicates adjusting it\n\n        This function converts ParseResult into dict\"\"\"\n\n        if pr.params:\n            lgr.warning(\"ParseResults contains params %r, which will be ignored\"\n                        % (pr.params,))\n\n        hostname_port = pr.netloc.split('@')[-1]\n        is_ipv6 = hostname_port.count(':') >= 2\n        # can't use just pr._asdict since we care to ask those properties\n        # such as .port , .hostname etc\n        # Forcing '' instead of None since those properties (.hostname), .password,\n        # .username return None if not available and we decided to uniformize\n        if is_ipv6:\n            rem = re.match('\\[(?P<hostname>.*)\\]:(?P<port>\\d+)', hostname_port)\n            if rem:\n                hostname, port = rem.groups()\n                port = int(port)\n            else:\n                hostname, port = hostname_port, ''\n\n            def _getattr(pr, f):\n                \"\"\"Helper for custom handling in case of ipv6 addresses which blows\n                stock ParseResults logic\"\"\"\n                if f == 'port':\n                    # for now not supported at all, so\n                    return port\n                elif f == 'hostname':\n                    return hostname\n                else:\n                    return getattr(pr, f)\n        else:\n            _getattr = getattr\n\n        return {f: (_getattr(pr, f) or '') for f in cls._FIELDS}\n\n    #\n    # Access helpers\n    #\n\n    def _parse_qs(self, s, auto_delist=True):\n        \"\"\"Helper around parse_qs to strip unneeded 'list'ing etc and return a dict of key=values\"\"\"\n        if not s:\n            return {}\n        out = OrderedDict(parse_qsl(s, 1))\n        if not auto_delist:\n            return out\n        for k in out:\n            v = out[k]\n            if isinstance(v, list) and len(v) == 1:\n                v = v[0]\n                out[k] = None if v == '' else v\n        return out\n\n    @property\n    def query_dict(self):\n        return self._parse_qs(self.query)\n\n    @property\n    def fragment_dict(self):\n        return self._parse_qs(self.fragment)\n\n    @property\n    def localpath(self):\n        if self.scheme != 'file':\n            raise ValueError(\n                \"Non 'file://' URL cannot be resolved to a local path\")\n        hostname = self.hostname\n        if not (hostname in (None, '', 'localhost', '::1') \\\n                or hostname.startswith('127.')):\n            raise ValueError(\"file:// URL does not point to 'localhost'\")\n        return self.path\n\n\n\nclass PathRI(RI):\n    \"\"\"RI pointing to a (local) file/directory\"\"\"\n    def as_str(self):\n        return self.path\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        return dict(path=url_str)\n\n    @property\n    def localpath(self):\n        return self.path\n\n\nclass RegexBasedURLMixin(object):\n    \"\"\"Base class for URLs which we could simple parse using regular expressions\"\"\"\n\n    _REGEX = None\n\n    # not used ATM but possible ;)\n    # @classmethod\n    # def is_str_matches(cls, url_str):\n    #     return bool(cls._REGEX.match(url_str))\n\n    @classmethod\n    def _str_to_fields(cls, url_str):\n        re_match = cls._REGEX.match(url_str)\n        if not re_match:\n            # TODO: custom error?\n            raise ValueError(\n                \"Possibly incorrectly determined string %r correspond to %s address\"\n                \" -- it failed matching regex. Dunno how to handle. Contact developers\"\n                % (cls, url_str,)\n            )\n        fields = cls._get_blank_fields()\n        fields.update({k: v for k, v in iteritems(re_match.groupdict()) if v})\n        cls._normalize_fields(fields)\n        return fields\n\n    @classmethod\n    def _normalize_fields(self, fields):\n        \"\"\"Helper to be ran if any of the fields need to be normalized after parsing\"\"\"\n        pass\n\n\nclass SSHRI(RI, RegexBasedURLMixin):\n    \"\"\"RI pointing to a remote location reachable via SSH\"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'username',\n        'hostname',\n        'port',\n    )\n\n    _REGEX = re.compile(r'((?P<username>\\S*)@)?(?P<hostname>[^:]+)(\\:(?P<path>.*))?$')\n\n    @classmethod\n    def _normalize_fields(cls, fields):\n        if fields['path'] and fields['path'].startswith('//'):\n            # Let's normalize for now to avoid multiple leading slashes\n            fields['path'] = '/' + fields['path'].lstrip('/')\n        # escape path so we have direct representation of the path to work with\n        fields['path'] = unescape_ssh_path(fields['path'])\n\n    def as_str(self):\n        fields = self.fields  # copy so we could escape symbols\n        url_fmt = '{hostname}'\n        if fields['username']:\n            url_fmt = \"{username}@\" + url_fmt\n        if fields['path']:\n            url_fmt += ':{path}'\n        fields['path'] = escape_ssh_path(fields['path'])\n        return url_fmt.format(**fields)\n\n    # TODO:\n    # we can \"support\" localhost:path as localpaths\n\n\nclass DataLadRI(RI, RegexBasedURLMixin):\n    \"\"\"RI pointing to datasets within central DataLad super-dataset\"\"\"\n\n    _FIELDS = RI._FIELDS + (\n        'remote',\n    )\n\n    # For now or forever we don't deal with any fragments or other special stuff\n    _REGEX = re.compile(r'//(?P<remote>[^\\s/]*)/(?P<path>.*)$')\n\n    # do they need to be normalized??? loosing track ...\n\n    def as_str(self):\n        return \"//{remote}/{path}\".format(**self._fields)\n\n    def as_git_url(self):\n        \"\"\"Dereference /// into original URLs which could be used by git for cloning\n\n        Returns\n        -------\n        str\n          URL string to reference the DataLadRI from its /// form\n        \"\"\"\n        if self.remote:\n            raise NotImplementedError(\"not supported ATM to reference additional remotes\")\n        return \"{}{}\".format(DATASETS_TOPURL, self.path)\n\n\ndef _split_colon(s, maxsplit=1):\n    \"\"\"Split on unescaped colon\"\"\"\n    return re.compile(r'(?<!\\\\):').split(s, maxsplit=maxsplit)\n\n# \\ should be first to deal with\n_SSH_ESCAPED_CHARACTERS = '\\\\#&;`|*?~<>^()[]{}$\\'\" '\n\n\n# TODO: RF using re.sub\ndef escape_ssh_path(path):\n    \"\"\"Escape all special characters present in the path\"\"\"\n    for c in _SSH_ESCAPED_CHARACTERS:\n        if c in path:\n            path = path.replace(c, '\\\\' + c)\n    return path\n\n\ndef unescape_ssh_path(path):\n    \"\"\"Un-escape all special characters present in the path\"\"\"\n    for c in _SSH_ESCAPED_CHARACTERS[::-1]:\n        if c in path:\n            path = path.replace('\\\\' + c, c)\n    return path\n\n\ndef parse_url_opts(url):\n    \"\"\"Given a string with url-style query, split into content before # and options as dict\"\"\"\n    url = URL(url)\n    # we need to filter out query and fragment to get the base url\n    fields = url.fields\n    fields.pop('query')\n    fields.pop('fragment')\n    opts = url.query_dict\n    return str(URL(**fields)), opts\n\n\n# TODO: should we just define URL.good_for_git or smth like that? ;)\n# although git also understands regular paths\ndef is_url(ri):\n    \"\"\"Returns whether argument is a resource identifier what datalad should treat as a URL\n\n    This includes ssh \"urls\" which git understands.\n\n    Parameters\n    ----------\n    ri : str or RI\n      The resource identifier (as a string or RI) to \"analyze\"\n    \"\"\"\n    if not isinstance(ri, RI):\n        try:\n            ri = RI(ri)\n        except:\n            return False\n    return isinstance(ri, (URL, SSHRI))\n\n\n# TODO: RF to remove duplication\ndef is_datalad_compat_ri(ri):\n    \"\"\"Returns whether argument is a resource identifier what datalad should treat as a URL\n\n    including its own DataLadRI\n    \"\"\"\n    if not isinstance(ri, RI):\n        try:\n            ri = RI(ri)\n        except:\n            return False\n    return isinstance(ri, (URL, SSHRI, DataLadRI))\n\n\n# TODO: better name? additionally may be move to SSHRI.is_valid() or sth.\ndef is_ssh(ri):\n    \"\"\"helper to determine, whether `ri` requires an SSH connection\n\n    Parameters\n    ----------\n    ri: str or RI\n\n    Returns\n    -------\n    bool\n    \"\"\"\n\n    # not exactly fitting the doc, but we actually can deal not necessarily with\n    # string or RI only, but with everything RI itself can deal with:\n    _ri = RI(ri) if not isinstance(ri, RI) else ri\n\n    return isinstance(_ri, SSHRI) or \\\n           (isinstance(_ri, URL) and _ri.scheme == 'ssh')\n\n\n#### windows workaround ###\n# TODO: There should be a better way\ndef get_local_file_url(fname):\n    \"\"\"Return OS specific URL pointing to a local file\n\n    Parameters\n    ----------\n    fname : string\n        Filename.  If not absolute, abspath is used\n    \"\"\"\n    fname = fname if isabs(fname) else abspath(fname)\n    if on_windows:\n        fname_rep = fname.replace('\\\\', '/')\n        furl = \"file:///%s\" % urlquote(fname_rep)\n        lgr.debug(\"Replaced '\\\\' in file\\'s url: %s\" % furl)\n    else:\n        # TODO:  need to fix for all the encoding etc\n        furl = str(URL(scheme='file', path=fname))\n    return furl"}
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:** Filename extraction from Content-Disposition header fails with varied casing

**Description:**
When handling HTTP responses, extracting the filename from the `Content-Disposition` header fails if the casing of the parameters varies. For example, using "Filename" instead of "filename" results in the filename not being retrieved correctly.

**Example Code:**
```python
headers = {
    'Content-Disposition': 'attachment; Filename="Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip"'
}

filename = get_response_disposition_filename(headers.get('Content-Disposition'))
print(filename)  # Expected: 'Glasser_et_al_2016_HCP_MMP1.0_RVVG.zip'
```

**Expected Behavior:**
The function `get_response_disposition_filename` should correctly extract and return the filename regardless of the casing of the `Content-Disposition` parameters.

**Actual Behavior:**
When the `Content-Disposition` header contains parameters with different casing (e.g., "Filename" instead of "filename"), the function fails to extract the filename and returns `None`.

[/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/support/network.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.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##

import calendar
import email.utils
import gzip
import os
import re
import shutil
import time
import iso8601

from collections import OrderedDict
from os.path import abspath, isabs

from six import string_types
from six import iteritems
from six.moves.urllib.request import urlopen, Request
from six.moves.urllib.parse import quote as urlquote, unquote as urlunquote
from six.moves.urllib.parse import urljoin, urlparse, urlsplit, urlunsplit, urlunparse, ParseResult
from six.moves.urllib.parse import parse_qsl
from six.moves.urllib.parse import urlencode
from six.moves.urllib.error import URLError

from datalad.utils import on_windows
from datalad.consts import DATASETS_TOPURL

from datalad.utils import auto_repr
# TODO not sure what needs to use `six` here yet
import requests

import logging
lgr = logging.getLogger('datalad.network')


def get_response_disposition_filename(s):
    """Given a string s as from HTTP Content-Disposition field in the response
    return possibly present filename if any
    """
    if not s:
        return None
    # If the response has Content-Disposition, try to get filename from it
    cd = dict(map(
        lambda x: x.strip().split('=') if '=' in x else (x.strip(),''),
        s.split(';')))
    if 'filename' in cd:
        filename = cd['filename'].strip("\"'")
        return filename
    return None


def get_url_disposition_filename(url, headers=None):
    """Get filename as possibly provided by the server in Content-Disposition
    """
    if headers is None:
        request = Request(url)
        r = retry_urlopen(request)
        # things are different in requests
        if 'requests.' in str(r.__class__):
            headers = r.headers
        else:
            headers = r.info()
    else:
        r = None
    try:
        return get_response_disposition_filename(headers.get('Content-Disposition', ''))
    finally:
        if r:
            r.close()


def get_url_straight_filename(url, strip=[], allowdir=False):
    """Get file/dir name of the last path component of the URL

    Parameters
    ----------
    strip: list, optional
      If provided, listed names will not be considered and their
      parent directory will be selected
    allowdir: bool, optional
      If url points to a "directory" (ends with /), empty string
      would be returned unless allowdir is True, in which case the
      name of the directory would be returned
    """
    path = urlunquote(urlsplit(url).path)
    path_parts = path.split('/')

    if allowdir:
        # strip empty ones
        while len(path_parts) > 1 and not path_parts[-1]:
            path_parts = path_parts[:-1]

    if strip:
        while path_parts and path_parts[-1] in strip:
            path_parts = path_parts[:-1]

    if path_parts:
        return path_parts[-1]
    else:
        return None


def get_url_filename(url, headers=None, strip=[]):
    """Get filename from the url, first consulting server about Content-Disposition
    """
    filename = get_url_disposition_filename(url, headers)
    if filename:
        return filename
    return get_url_straight_filename(url, strip=[])


def get_url_response_stamp(url, response_info):
    size, mtime = None, None
    if 'Content-length' in response_info:
        size = int(response_info['Content-length'])
    if 'Last-modified' in response_info:
        mtime = calendar.timegm(email.utils.parsedate(
            response_info['Last-modified']))
    return dict(size=size, mtime=mtime, url=url)


def get_tld(url):
    """Return top level domain from a url

    Parameters
    ----------
    url : str
    """
    # maybe use this instead to be safe:  https://pypi.python.org/pypi/tld
    if not url.strip():
        raise ValueError("Empty URL has no TLD")
    rec = urlsplit(url)
    if not rec.netloc:
        if not rec.scheme:
            # There were no scheme provided thus netloc was empty -- must have been a simple 'path like'
            return url.split('/', 1)[0]
        else:
            raise ValueError("It seems that only the scheme was provided without the net location/TLD")
    return rec.netloc


from email.utils import parsedate_tz, mktime_tz
def rfc2822_to_epoch(datestr):
    """Given rfc2822 date/time format, return seconds since epoch"""
    return mktime_tz(parsedate_tz(datestr))


import calendar
def iso8601_to_epoch(datestr):
    """Given ISO 8601 date/time format, return in seconds since epoch

    iso8601 is used to parse properly the time zone information, which
    can't be parsed with standard datetime strptime
    """
    return calendar.timegm(iso8601.parse_date(datestr).timetuple())


def __urlopen_requests(url, header_vals=None):
    # XXX Workaround for now for ... broken code
    if isinstance(url, Request):
        url = url.get_full_url()
    return requests.Session().get(url)


def retry_urlopen(url, retries=3):
    for t in range(retries):
        try:
            return __urlopen_requests(url)
        except URLError as e:
            lgr.warn("Received exception while reading %s: %s" % (url, e))
            if t == retries - 1:
                # if we have reached allowed number of retries -- reraise
                raise


def is_url_quoted(url):
    """Return either URL looks being already quoted
    """
    try:
        url_ = urlunquote(url)
        return url != url_
    except:  # problem with unquoting -- then it must be wasn't quoted (correctly)
        return False


def same_website(url_rec, u_rec):
    """Decide either a link leads to external site

    Parameters
    ----------
    url_rec: ParseResult
      record for original url
    u_rec: ParseResult
      record for new url
    """
    if isinstance(url_rec, string_types):
        url_rec = urlparse(url_rec)
    if isinstance(u_rec, string_types):
        u_rec = urlparse(u_rec)
    return (url_rec.netloc == u_rec.netloc)
    # todo: collect more of sample cases.
    # disabled below check while working on ratholeradio, since links
    # could go to the parent and that is ok.  Figure out when it was
    # desired not to go to the parent -- we might need explicit option
    # and u_rec.path.startswith(url_rec.path)):


def dlurljoin(u_path, url):
    url_rec = urlparse(url)  # probably duplicating parsing :-/ TODO
    if url_rec.scheme:
        # independent full url, so just return it
        return url
    if u_path.endswith('/'):  # should here be also a scheme use?
        if url.startswith('/'):  # jump to the root
            u_path_rec = urlparse(u_path)
            return urljoin(urlunparse(
                (u_path_rec.scheme, u_path_rec.netloc, '', '', '', '')), url)
        else:
            return os.path.join(u_path, url)
    # TODO: recall where all this dirname came from and bring into the test
    return urljoin(os.path.dirname(u_path) + '/', url)


# TODO should it be a node maybe?
class SimpleURLStamper(object):
    """Gets a simple stamp about the URL: {url, time, size} of whatever was provided in the header
    """
    def __init__(self, mode='full'):
        self.mode = mode

    def __call__(self, url):
        # Extracted from above madness
        # TODO: add mode alike to 'relaxed' where we would not
        # care about content-disposition filename
        # http://stackoverflow.com/questions/862173/how-to-download-a-file-using-python-in-a-smarter-way
        request = Request(url)

        # No traffic compression since we do not know how to identify
        # exactly either it has to be decompressed
        # request.add_header('Accept-encoding', 'gzip,deflate')
        #
        # TODO: think about stamping etc -- we seems to be redoing
        # what git-annex does for us already... not really
        r = retry_urlopen(request)
        try:
            r_info = r.info()
            r_stamp = get_url_response_stamp(url, r_info)

            return dict(mtime=r_stamp['mtime'], size=r_stamp['size'], url=url)
        finally:
            r.close()


# TODO:  make it consistent/clear at what stage % encoding/decoding happens!
# now it is a mix!

#
# Useful functionality in requests.models
#  utils.requote_uri -- quote/unquote cycle to guarantee consistent appearance
#  RequestEncodingMixin._encode_params -- Will successfully encode parameters when passed as a dict or a list of ...
#  PreparedRequest().prepare_url(url, params) -- nicely cares about url encodings etc
#

def _guess_ri_cls(ri):
    """Factory function which would determine which type of a ri a provided string is"""
    # We assume that it is a URL and parse it. Depending on the result
    # we might decide that it was something else ;)
    fields = URL._pr_to_fields(urlparse(ri))
    lgr.log(5, "Parsed ri %s into fields %s" % (ri, fields))

    type_ = 'url'
    # Special treatments
    # file:///path should stay file:
    if fields['scheme'] and fields['scheme'] not in {'file'} \
            and not fields['hostname']:
        # dl+archive:... or just for ssh   hostname:path/p1
        if '+' not in fields['scheme']:
            type_ = 'ssh'
            lgr.log(5, "Assuming ssh style ri, adjusted: %s" % (fields,))

    if not fields['scheme'] and not fields['hostname']:
        parts = _split_colon(ri)
        if fields['path'] and '@' in fields['path'] or len(parts) > 1:
            # user@host:path/sp1
            # or host_name: (hence parts check)
            # TODO: we need a regex to catch those really, parts check is not suff
            type_ = 'ssh'
        elif ri.startswith('//'):
            # e.g. // or ///path
            type_ = 'datalad'
        else:
            type_ = 'file'

    if not fields['scheme'] and fields['hostname']:
        # e.g. //a/path
        type_ = 'datalad'

    TYPES = {
        'url': URL,
        'ssh':  SSHRI,
        'file': PathRI,
        'datalad': DataLadRI
    }
    cls = TYPES[type_]
    # just parse the ri according to regex matchint ssh "ri" specs
    lgr.log(5, "Detected %s ri" % type_)
    return cls


class RI(object):
    """Resource Identifier - base class and a factory for URL, SSHRI, etc

    Intended to be a R/O object (i.e. no fields should be changed in-place).
    Subclasses define specific collections of fields they care about in _FIELDS
    class variable.
    The idea is that this class should help to break apart a URL, while being
    able to rebuild itself into a string representation for reuse

    `RI` could be used as factory, whenever type of the resource is unknown and
    must be guessed from the string representation.  One of the subclasses will be
    provided as output, e.g.

    >>> RI('http://example.com')
    URL(hostname='example.com', scheme='http')
    >>> RI('example.com:path')
    SSHRI(hostname='example.com', path='path')
    """

    # All of the subclasses will provide path
    _FIELDS = (
        'path',
    )

    __slots__ = _FIELDS + ('_fields', '_str')

    def __new__(cls, ri=None, **kwargs):
        """Used as a possible factory for known RI types

        Returns
        -------
        RI
           uninitialized RI object of appropriate class with _str
           set to string representation if was provided

        """
        if cls is RI and ri is not None:
            # RI class was used as a factory
            cls = _guess_ri_cls(ri)

        ri_obj = super(RI, cls).__new__(cls)
        # Store internally original str
        ri_obj._str = ri
        return ri_obj

    def __init__(self, ri=None, **fields):
        """
        Parameters
        ----------
        ri: str, optional
          String version of a resource specific for this class.  If you would like
          a type of the resource be deduced, use RI(ri)
        **fields: dict, optional
          The values for the fields defined in _FIELDS class variable.
        """
        if ri and (bool(ri) == bool(fields)):
            raise ValueError(
                "Specify either ri or breakdown from the fields, not both. "
                "Got ri=%r, fields=%r" % (ri, fields))

        self._fields = self._get_blank_fields()
        if ri is not None:
            fields = self._str_to_fields(ri)
        self._set_from_fields(**fields)

        # If was initialized from a string representation
        if self._str is not None:
            # well -- some ris might not unparse identically back
            # strictly speaking, but let's assume they do
            ri_ = self.as_str()
            if ri != ri_:
                lgr.warning("Parsed version of %s %r differs from original %r",
                            self.__class__.__name__, ri_, ri)

    @classmethod
    def _get_blank_fields(cls, **fields):
        return OrderedDict(((f, fields.get(f, '')) for f in cls._FIELDS))

    @property
    def fields(self):
        """Returns shallow copy of fields to ease manipulations"""
        return self._fields.copy()

    def __repr__(self):
        # since auto_repr doesn't support "non-0" values atm
        return "%s(%s)" % (
            self.__class__.__name__,
            ", ".join(["%s=%r" % (k, v)
                       for k, v in sorted(self._fields.items())
                       if v]))

    # Lazily evaluated if _str was not set
    def __str__(self):
        if self._str is None:
            self._str = self.as_str()
        return self._str

    @classmethod
    def from_str(cls, ri_str):
        obj = cls(**cls._str_to_fields(ri_str))
        obj._str = ri_str
        return obj

    @property
    def localpath(self):
        # by default RIs point to remote locations
        raise ValueError("%s points to remote location" % self)

    # Apparently doesn't quite play nicely with multiple inheritence for MixIn'
    # of regexp based URLs
    #@abstractmethod
    #@classmethod
    #def _str_to_fields(cls, ri_str):
    #    raise NotImplementedError

    #
    # If any field is specified, URL is not considered 'False', i.e.
    # non-existing, although may be we could/shout omit having only
    # scheme or port specified since it doesn't point to any useful
    # location
    #

    def __nonzero__(self):
        fields = self._fields
        return any(fields.values())

    # for PY3
    __bool__ = __nonzero__

    #
    # Helpers to deal with internal structures and conversions
    #

    def _set_from_fields(self, **fields):
        unknown_fields = set(fields).difference(self._FIELDS)
        if unknown_fields:
            raise ValueError("Do not know about %s. Known fields for %s are: %s"
                             % (unknown_fields, self.__class__, self._FIELDS))

        # encode dicts for query or fragment into
        for f in {'query', 'fragment'}:
            v = fields.get(f)
            if isinstance(v, dict):

                ev = urlencode(v)
                # / is reserved char within query
                if f == 'fragment' and '%2F' not in str(v):
                    # but seems to be ok'ish within the fragment which is
                    # the last element of URI and anyways used only by the
                    # client (i.e. by us here if used to compose the URL)
                    # so let's return / back for clarity if there were no
                    # awkward %2F to startswith
                    ev = ev.replace('%2F', '/')
                fields[f] = ev

        self._fields.update(fields)

    #
    # Quick comparators
    #

    def __eq__(self, other):
        if not isinstance(other, RI):
            other = RI(other)
        return isinstance(other, self.__class__) and dict(other._fields) == dict(self._fields)

    def __ne__(self, other):
        return not (self == other)

    def __getattribute__(self, item):
        if item.startswith('_') or item not in self._FIELDS:
            return super(RI, self).__getattribute__(item)
        else:
            return self._fields[item]


class URL(RI):
    """Universal resource locator

    Although largely decorating urlparse.ParseResult, it
    - doesn't mandate providing all parts of the URL
    - doesn't require netloc but rather asks for separate username, password, and hostname
    """

    _FIELDS = RI._FIELDS + (
        'scheme',
        'username',
        'password',
        'hostname', 'port',
        'query',
        'fragment',
    )

    def as_str(self):
        """Render URL as a string"""
        return urlunparse(self.to_pr())

    @classmethod
    def _str_to_fields(cls, url_str):
        fields = URL._pr_to_fields(urlparse(url_str))
        fields['path'] = urlunquote(fields['path'])
        return fields

    def to_pr(self):
        """Convert URL to urlparse.ParseResults namedtuple"""
        return self._fields_to_pr(self._fields)

    @classmethod
    def _fields_to_pr(cls, fields):
        """Recompose back fields dict to ParseResult"""
        netloc = fields['username'] or ''
        if fields['password']:
            netloc += ':' + fields['password']
        if netloc:
            netloc += '@'
        netloc += fields['hostname']
        if fields['port']:
            if fields['hostname'].count(':') >= 2:
                # ipv6 -- need to enclose in []
                netloc = '[%s]:%s' % (netloc, fields['port'])
            else:
                netloc += ':%s' % fields['port']

        pr_fields = {
            f: fields[f]
            for f in cls._FIELDS
            if f not in ('hostname', 'password', 'username', 'port')
        }
        pr_fields['netloc'] = netloc
        pr_fields['params'] = ''
        # We need to quote the path
        pr_fields['path'] = urlquote(pr_fields['path'])
        # TODO: figure out what to do with query/fragment... one step at a time
        return ParseResult(**pr_fields)

    @classmethod
    def _pr_to_fields(cls, pr):
        """ParseResult is a tuple so immutable, which complicates adjusting it

        This function converts ParseResult into dict"""

        if pr.params:
            lgr.warning("ParseResults contains params %r, which will be ignored"
                        % (pr.params,))

        hostname_port = pr.netloc.split('@')[-1]
        is_ipv6 = hostname_port.count(':') >= 2
        # can't use just pr._asdict since we care to ask those properties
        # such as .port , .hostname etc
        # Forcing '' instead of None since those properties (.hostname), .password,
        # .username return None if not available and we decided to uniformize
        if is_ipv6:
            rem = re.match('\[(?P<hostname>.*)\]:(?P<port>\d+)', hostname_port)
            if rem:
                hostname, port = rem.groups()
                port = int(port)
            else:
                hostname, port = hostname_port, ''

            def _getattr(pr, f):
                """Helper for custom handling in case of ipv6 addresses which blows
                stock ParseResults logic"""
                if f == 'port':
                    # for now not supported at all, so
                    return port
                elif f == 'hostname':
                    return hostname
                else:
                    return getattr(pr, f)
        else:
            _getattr = getattr

        return {f: (_getattr(pr, f) or '') for f in cls._FIELDS}

    #
    # Access helpers
    #

    def _parse_qs(self, s, auto_delist=True):
        """Helper around parse_qs to strip unneeded 'list'ing etc and return a dict of key=values"""
        if not s:
            return {}
        out = OrderedDict(parse_qsl(s, 1))
        if not auto_delist:
            return out
        for k in out:
            v = out[k]
            if isinstance(v, list) and len(v) == 1:
                v = v[0]
                out[k] = None if v == '' else v
        return out

    @property
    def query_dict(self):
        return self._parse_qs(self.query)

    @property
    def fragment_dict(self):
        return self._parse_qs(self.fragment)

    @property
    def localpath(self):
        if self.scheme != 'file':
            raise ValueError(
                "Non 'file://' URL cannot be resolved to a local path")
        hostname = self.hostname
        if not (hostname in (None, '', 'localhost', '::1') \
                or hostname.startswith('127.')):
            raise ValueError("file:// URL does not point to 'localhost'")
        return self.path



class PathRI(RI):
    """RI pointing to a (local) file/directory"""
    def as_str(self):
        return self.path

    @classmethod
    def _str_to_fields(cls, url_str):
        return dict(path=url_str)

    @property
    def localpath(self):
        return self.path


class RegexBasedURLMixin(object):
    """Base class for URLs which we could simple parse using regular expressions"""

    _REGEX = None

    # not used ATM but possible ;)
    # @classmethod
    # def is_str_matches(cls, url_str):
    #     return bool(cls._REGEX.match(url_str))

    @classmethod
    def _str_to_fields(cls, url_str):
        re_match = cls._REGEX.match(url_str)
        if not re_match:
            # TODO: custom error?
            raise ValueError(
                "Possibly incorrectly determined string %r correspond to %s address"
                " -- it failed matching regex. Dunno how to handle. Contact developers"
                % (cls, url_str,)
            )
        fields = cls._get_blank_fields()
        fields.update({k: v for k, v in iteritems(re_match.groupdict()) if v})
        cls._normalize_fields(fields)
        return fields

    @classmethod
    def _normalize_fields(self, fields):
        """Helper to be ran if any of the fields need to be normalized after parsing"""
        pass


class SSHRI(RI, RegexBasedURLMixin):
    """RI pointing to a remote location reachable via SSH"""

    _FIELDS = RI._FIELDS + (
        'username',
        'hostname',
        'port',
    )

    _REGEX = re.compile(r'((?P<username>\S*)@)?(?P<hostname>[^:]+)(\:(?P<path>.*))?$')

    @classmethod
    def _normalize_fields(cls, fields):
        if fields['path'] and fields['path'].startswith('//'):
            # Let's normalize for now to avoid multiple leading slashes
            fields['path'] = '/' + fields['path'].lstrip('/')
        # escape path so we have direct representation of the path to work with
        fields['path'] = unescape_ssh_path(fields['path'])

    def as_str(self):
        fields = self.fields  # copy so we could escape symbols
        url_fmt = '{hostname}'
        if fields['username']:
            url_fmt = "{username}@" + url_fmt
        if fields['path']:
            url_fmt += ':{path}'
        fields['path'] = escape_ssh_path(fields['path'])
        return url_fmt.format(**fields)

    # TODO:
    # we can "support" localhost:path as localpaths


class DataLadRI(RI, RegexBasedURLMixin):
    """RI pointing to datasets within central DataLad super-dataset"""

    _FIELDS = RI._FIELDS + (
        'remote',
    )

    # For now or forever we don't deal with any fragments or other special stuff
    _REGEX = re.compile(r'//(?P<remote>[^\s/]*)/(?P<path>.*)$')

    # do they need to be normalized??? loosing track ...

    def as_str(self):
        return "//{remote}/{path}".format(**self._fields)

    def as_git_url(self):
        """Dereference /// into original URLs which could be used by git for cloning

        Returns
        -------
        str
          URL string to reference the DataLadRI from its /// form
        """
        if self.remote:
            raise NotImplementedError("not supported ATM to reference additional remotes")
        return "{}{}".format(DATASETS_TOPURL, self.path)


def _split_colon(s, maxsplit=1):
    """Split on unescaped colon"""
    return re.compile(r'(?<!\\):').split(s, maxsplit=maxsplit)

# \ should be first to deal with
_SSH_ESCAPED_CHARACTERS = '\\#&;`|*?~<>^()[]{}$\'" '


# TODO: RF using re.sub
def escape_ssh_path(path):
    """Escape all special characters present in the path"""
    for c in _SSH_ESCAPED_CHARACTERS:
        if c in path:
            path = path.replace(c, '\\' + c)
    return path


def unescape_ssh_path(path):
    """Un-escape all special characters present in the path"""
    for c in _SSH_ESCAPED_CHARACTERS[::-1]:
        if c in path:
            path = path.replace('\\' + c, c)
    return path


def parse_url_opts(url):
    """Given a string with url-style query, split into content before # and options as dict"""
    url = URL(url)
    # we need to filter out query and fragment to get the base url
    fields = url.fields
    fields.pop('query')
    fields.pop('fragment')
    opts = url.query_dict
    return str(URL(**fields)), opts


# TODO: should we just define URL.good_for_git or smth like that? ;)
# although git also understands regular paths
def is_url(ri):
    """Returns whether argument is a resource identifier what datalad should treat as a URL

    This includes ssh "urls" which git understands.

    Parameters
    ----------
    ri : str or RI
      The resource identifier (as a string or RI) to "analyze"
    """
    if not isinstance(ri, RI):
        try:
            ri = RI(ri)
        except:
            return False
    return isinstance(ri, (URL, SSHRI))


# TODO: RF to remove duplication
def is_datalad_compat_ri(ri):
    """Returns whether argument is a resource identifier what datalad should treat as a URL

    including its own DataLadRI
    """
    if not isinstance(ri, RI):
        try:
            ri = RI(ri)
        except:
            return False
    return isinstance(ri, (URL, SSHRI, DataLadRI))


# TODO: better name? additionally may be move to SSHRI.is_valid() or sth.
def is_ssh(ri):
    """helper to determine, whether `ri` requires an SSH connection

    Parameters
    ----------
    ri: str or RI

    Returns
    -------
    bool
    """

    # not exactly fitting the doc, but we actually can deal not necessarily with
    # string or RI only, but with everything RI itself can deal with:
    _ri = RI(ri) if not isinstance(ri, RI) else ri

    return isinstance(_ri, SSHRI) or \
           (isinstance(_ri, URL) and _ri.scheme == 'ssh')


#### windows workaround ###
# TODO: There should be a better way
def get_local_file_url(fname):
    """Return OS specific URL pointing to a local file

    Parameters
    ----------
    fname : string
        Filename.  If not absolute, abspath is used
    """
    fname = fname if isabs(fname) else abspath(fname)
    if on_windows:
        fname_rep = fname.replace('\\', '/')
        furl = "file:///%s" % urlquote(fname_rep)
        lgr.debug("Replaced '\\' in file\'s url: %s" % furl)
    else:
        # TODO:  need to fix for all the encoding etc
        furl = str(URL(scheme='file', path=fname))
    return furl
```
--- 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