Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/tests/test_utils.py
| Show All 29 Lines | def test_download_fail_to_download(tmp_path, requests_mock): | ||||
| requests_mock.get(url, status_code=status_code) | requests_mock.get(url, status_code=status_code) | ||||
| with pytest.raises(ValueError) as e: | with pytest.raises(ValueError) as e: | ||||
| download(url, tmp_path) | download(url, tmp_path) | ||||
| assert e.value.args[0] == "Fail to query '%s'. Reason: %s" % (url, status_code) | assert e.value.args[0] == "Fail to query '%s'. Reason: %s" % (url, status_code) | ||||
| @pytest.mark.fs | _filename = "requests-0.0.1.tar.gz" | ||||
| def test_download_ok(tmp_path, requests_mock): | _data = "this is something" | ||||
| """Download without issue should provide filename and hashes""" | |||||
| filename = "requests-0.0.1.tar.gz" | |||||
| url = "https://pypi.org/pypi/requests/%s" % filename | |||||
| data = "this is something" | |||||
| requests_mock.get(url, text=data, headers={"content-length": str(len(data))}) | |||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | |||||
| def _check_download_ok(url, dest, filename=_filename, hashes=None): | |||||
| actual_filepath, actual_hashes = download(url, dest, hashes=hashes) | |||||
| actual_filename = os.path.basename(actual_filepath) | actual_filename = os.path.basename(actual_filepath) | ||||
| assert actual_filename == filename | assert actual_filename == filename | ||||
| assert actual_hashes["length"] == len(data) | assert actual_hashes["length"] == len(_data) | ||||
| assert ( | assert ( | ||||
| actual_hashes["checksums"]["sha1"] == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | actual_hashes["checksums"]["sha1"] == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | ||||
| ) # noqa | ) | ||||
| assert ( | assert ( | ||||
| actual_hashes["checksums"]["sha256"] | actual_hashes["checksums"]["sha256"] | ||||
| == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | ||||
| ) | ) | ||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_download_ok_no_header(tmp_path, requests_mock): | def test_download_ok(tmp_path, requests_mock): | ||||
| """Download without issue should provide filename and hashes""" | """Download without issue should provide filename and hashes""" | ||||
| filename = "requests-0.0.1.tar.gz" | url = f"https://pypi.org/pypi/requests/{_filename}" | ||||
| url = "https://pypi.org/pypi/requests/%s" % filename | requests_mock.get(url, text=_data, headers={"content-length": str(len(_data))}) | ||||
| data = "this is something" | _check_download_ok(url, dest=str(tmp_path)) | ||||
| requests_mock.get(url, text=data) # no header information | |||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | |||||
| actual_filename = os.path.basename(actual_filepath) | @pytest.mark.fs | ||||
| assert actual_filename == filename | def test_download_ok_no_header(tmp_path, requests_mock): | ||||
| assert actual_hashes["length"] == len(data) | """Download without issue should provide filename and hashes""" | ||||
| assert ( | url = f"https://pypi.org/pypi/requests/{_filename}" | ||||
| actual_hashes["checksums"]["sha1"] == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | requests_mock.get(url, text=_data) # no header information | ||||
| ) # noqa | _check_download_ok(url, dest=str(tmp_path)) | ||||
| assert ( | |||||
| actual_hashes["checksums"]["sha256"] | |||||
| == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | |||||
| ) | |||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_download_ok_with_hashes(tmp_path, requests_mock): | def test_download_ok_with_hashes(tmp_path, requests_mock): | ||||
| """Download without issue should provide filename and hashes""" | """Download without issue should provide filename and hashes""" | ||||
| filename = "requests-0.0.1.tar.gz" | url = f"https://pypi.org/pypi/requests/{_filename}" | ||||
| url = "https://pypi.org/pypi/requests/%s" % filename | requests_mock.get(url, text=_data, headers={"content-length": str(len(_data))}) | ||||
| data = "this is something" | |||||
| requests_mock.get(url, text=data, headers={"content-length": str(len(data))}) | |||||
| # good hashes for such file | # good hashes for such file | ||||
| good = { | good = { | ||||
| "sha1": "fdd1ce606a904b08c816ba84f3125f2af44d92b2", | "sha1": "fdd1ce606a904b08c816ba84f3125f2af44d92b2", | ||||
| "sha256": "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5", # noqa | "sha256": "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5", # noqa | ||||
| } | } | ||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path), hashes=good) | _check_download_ok(url, dest=str(tmp_path), hashes=good) | ||||
| actual_filename = os.path.basename(actual_filepath) | |||||
| assert actual_filename == filename | |||||
| assert actual_hashes["length"] == len(data) | |||||
| assert actual_hashes["checksums"]["sha1"] == good["sha1"] | |||||
| assert actual_hashes["checksums"]["sha256"] == good["sha256"] | |||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_download_fail_hashes_mismatch(tmp_path, requests_mock): | def test_download_fail_hashes_mismatch(tmp_path, requests_mock): | ||||
| """Mismatch hash after download should raise | """Mismatch hash after download should raise | ||||
| """ | """ | ||||
| filename = "requests-0.0.1.tar.gz" | url = f"https://pypi.org/pypi/requests/{_filename}" | ||||
| url = "https://pypi.org/pypi/requests/%s" % filename | requests_mock.get(url, text=_data, headers={"content-length": str(len(_data))}) | ||||
| data = "this is something" | |||||
| requests_mock.get(url, text=data, headers={"content-length": str(len(data))}) | |||||
| # good hashes for such file | # good hashes for such file | ||||
| good = { | good = { | ||||
| "sha1": "fdd1ce606a904b08c816ba84f3125f2af44d92b2", | "sha1": "fdd1ce606a904b08c816ba84f3125f2af44d92b2", | ||||
| "sha256": "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5", # noqa | "sha256": "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5", # noqa | ||||
| } | } | ||||
| for hash_algo in good.keys(): | for hash_algo in good.keys(): | ||||
| Show All 9 Lines | for hash_algo in good.keys(): | ||||
| with pytest.raises(ValueError, match=expected_msg): | with pytest.raises(ValueError, match=expected_msg): | ||||
| download(url, dest=str(tmp_path), hashes=expected_hashes) | download(url, dest=str(tmp_path), hashes=expected_hashes) | ||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_ftp_download_ok(tmp_path, mocker): | def test_ftp_download_ok(tmp_path, mocker): | ||||
| """Download without issue should provide filename and hashes""" | """Download without issue should provide filename and hashes""" | ||||
| filename = "requests-0.0.1.tar.gz" | url = f"ftp://pypi.org/pypi/requests/{_filename}" | ||||
| url = "ftp://pypi.org/pypi/requests/%s" % filename | |||||
| data = b"this is something" | |||||
| cm = MagicMock() | cm = MagicMock() | ||||
| cm.getstatus.return_value = 200 | cm.getstatus.return_value = 200 | ||||
| cm.read.side_effect = [data, b""] | cm.read.side_effect = [_data.encode(), b""] | ||||
| cm.__enter__.return_value = cm | cm.__enter__.return_value = cm | ||||
| mocker.patch("swh.loader.package.utils.urlopen").return_value = cm | mocker.patch("swh.loader.package.utils.urlopen").return_value = cm | ||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | _check_download_ok(url, dest=str(tmp_path)) | ||||
| actual_filename = os.path.basename(actual_filepath) | |||||
| assert actual_filename == filename | |||||
| assert actual_hashes["length"] == len(data) | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha1"] == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | |||||
| ) # noqa | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha256"] | |||||
| == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | |||||
| ) | |||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_ftp_download_ko(tmp_path, mocker): | def test_ftp_download_ko(tmp_path, mocker): | ||||
| """Download without issue should provide filename and hashes""" | """Download without issue should provide filename and hashes""" | ||||
| filename = "requests-0.0.1.tar.gz" | filename = "requests-0.0.1.tar.gz" | ||||
| url = "ftp://pypi.org/pypi/requests/%s" % filename | url = "ftp://pypi.org/pypi/requests/%s" % filename | ||||
| mocker.patch("swh.loader.package.utils.urlopen").side_effect = URLError("FTP error") | mocker.patch("swh.loader.package.utils.urlopen").side_effect = URLError("FTP error") | ||||
| with pytest.raises(URLError): | with pytest.raises(URLError): | ||||
| download(url, dest=str(tmp_path)) | download(url, dest=str(tmp_path)) | ||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_download_with_redirection(tmp_path, requests_mock): | def test_download_with_redirection(tmp_path, requests_mock): | ||||
| """Download with redirection should use the targeted URL to extract filename""" | """Download with redirection should use the targeted URL to extract filename""" | ||||
| url = "https://example.org/project/requests/download" | url = "https://example.org/project/requests/download" | ||||
| filename = "requests-0.0.1.tar.gz" | redirection_url = f"https://example.org/project/requests/files/{_filename}" | ||||
| redirection_url = f"https://example.org/project/requests/files/{filename}" | |||||
| data = "this is something" | |||||
| requests_mock.get(url, status_code=302, headers={"location": redirection_url}) | requests_mock.get(url, status_code=302, headers={"location": redirection_url}) | ||||
| requests_mock.get( | requests_mock.get( | ||||
| redirection_url, text=data, headers={"content-length": str(len(data))} | redirection_url, text=_data, headers={"content-length": str(len(_data))} | ||||
| ) | ) | ||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | _check_download_ok(url, dest=str(tmp_path)) | ||||
| actual_filename = os.path.basename(actual_filepath) | |||||
| assert actual_filename == filename | |||||
| assert actual_hashes["length"] == len(data) | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha1"] == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | |||||
| ) # noqa | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha256"] | |||||
| == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | |||||
| ) | |||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_download_filename_from_content_disposition(tmp_path, requests_mock): | @pytest.mark.parametrize( | ||||
| "filename", [f'"{_filename}"', _filename, '"filename with spaces.tar.gz"'] | |||||
| ) | |||||
| def test_download_filename_from_content_disposition(tmp_path, requests_mock, filename): | |||||
| """Filename should be extracted from content-disposition request header | """Filename should be extracted from content-disposition request header | ||||
| when available.""" | when available.""" | ||||
| url = "https://example.org/download/requests/tar.gz/v0.0.1" | url = "https://example.org/download/requests/tar.gz/v0.0.1" | ||||
| filename = "requests-0.0.1.tar.gz" | |||||
| data = "this is something" | |||||
| for fname in (f'"{filename}"', filename, '"filename with spaces.tar.gz"'): | |||||
| requests_mock.get( | requests_mock.get( | ||||
| url, | url, | ||||
| text=data, | text=_data, | ||||
| headers={ | headers={ | ||||
| "content-length": str(len(data)), | "content-length": str(len(_data)), | ||||
| "content-disposition": f"attachment; filename={fname}", | "content-disposition": f"attachment; filename={filename}", | ||||
| }, | }, | ||||
| ) | ) | ||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | _check_download_ok(url, dest=str(tmp_path), filename=filename.strip('"')) | ||||
| actual_filename = os.path.basename(actual_filepath) | |||||
| assert actual_filename == fname.strip('"') | |||||
| assert actual_hashes["length"] == len(data) | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha1"] | |||||
| == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | |||||
| ) # noqa | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha256"] | |||||
| == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | |||||
| ) | |||||
| @pytest.mark.fs | @pytest.mark.fs | ||||
| def test_download_utf8_filename_from_content_disposition(tmp_path, requests_mock): | @pytest.mark.parametrize("filename", ['"archive école.tar.gz"', "archive_école.tgz"]) | ||||
| def test_download_utf8_filename_from_content_disposition( | |||||
| tmp_path, requests_mock, filename | |||||
| ): | |||||
| """Filename should be extracted from content-disposition request header | """Filename should be extracted from content-disposition request header | ||||
| when available.""" | when available.""" | ||||
| url = "https://example.org/download/requests/tar.gz/v0.0.1" | url = "https://example.org/download/requests/tar.gz/v0.0.1" | ||||
| data = "this is something" | data = "this is something" | ||||
| for fname in ('"archive école.tar.gz"', "archive_école.tgz"): | |||||
| requests_mock.get( | requests_mock.get( | ||||
| url, | url, | ||||
| text=data, | text=data, | ||||
| headers={ | headers={ | ||||
| "content-length": str(len(data)), | "content-length": str(len(data)), | ||||
| "content-disposition": f"attachment; filename*=utf-8''{quote(fname)}", | "content-disposition": f"attachment; filename*=utf-8''{quote(filename)}", | ||||
| }, | }, | ||||
| ) | ) | ||||
| actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | _check_download_ok(url, dest=str(tmp_path), filename=filename.strip('"')) | ||||
| actual_filename = os.path.basename(actual_filepath) | |||||
| assert actual_filename == fname.strip('"') | |||||
| assert actual_hashes["length"] == len(data) | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha1"] | |||||
| == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | |||||
| ) # noqa | |||||
| assert ( | |||||
| actual_hashes["checksums"]["sha256"] | |||||
| == "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | |||||
| ) | |||||
| def test_api_info_failure(requests_mock): | def test_api_info_failure(requests_mock): | ||||
| """Failure to fetch info/release information should raise""" | """Failure to fetch info/release information should raise""" | ||||
| url = "https://pypi.org/pypi/requests/json" | url = "https://pypi.org/pypi/requests/json" | ||||
| status_code = 400 | status_code = 400 | ||||
| requests_mock.get(url, status_code=status_code) | requests_mock.get(url, status_code=status_code) | ||||
| Show All 22 Lines | |||||