diff --git a/test/test_download.py b/test/test_download.py new file mode 100644 index 0000000..0d9880f --- /dev/null +++ b/test/test_download.py @@ -0,0 +1,90 @@ +import os + +import pytest +import requests +import requests_mock + +import wxee.download +from wxee.exceptions import DownloadError + +TEST_URL = "https://amockurlfortestingwxee.biz" +TEST_OUT_DIR = os.path.join("test", "test_data") +TEST_CONTENT = "Wenn ist das Nunstück git und Slotermeyer? Ja! Beiherhund das Oder die Flipperwaldt gersput!" +CONTENT_LENGTH = len(TEST_CONTENT.encode("utf-8")) +BAD_CODES = [ + requests.codes.internal_server_error, + requests.codes.bad_gateway, + requests.codes.service_unavailable, + requests.codes.gateway_timeout, + requests.codes.too_many_requests, + requests.codes.request_timeout, + requests.codes.not_found, +] + + +def test_download_url_creates_file(): + """Test that the download_url function downloads a mock file with correct content.""" + with requests_mock.Mocker() as m: + m.get( + TEST_URL, text=TEST_CONTENT, headers={"content-length": str(CONTENT_LENGTH)} + ) + file = wxee.download._download_url(TEST_URL, TEST_OUT_DIR, progress=False) + + assert os.path.isfile(file) + + with open(file, "r") as result: + assert result.read() == TEST_CONTENT + + os.remove(file) + + +def test_download_url_warns_if_incomplete(): + """Test that the download_url function raises a warning when incomplete data is downloaded.""" + with requests_mock.Mocker() as m: + # Set an incorrect content length to trigger the incomplete download warning + m.get( + TEST_URL, + text=TEST_CONTENT, + headers={"content-length": str(CONTENT_LENGTH + 5)}, + ) + + with pytest.warns(UserWarning): + file = wxee.download._download_url( + TEST_URL, TEST_OUT_DIR, progress=False, max_attempts=1 + ) + + os.remove(file) + + +def test_download_url_fails_with_bad_status(): + """Test that the download_url function fails correctly with a bad status code.""" + with requests_mock.Mocker() as m: + for code in BAD_CODES: + m.get(TEST_URL, text=TEST_CONTENT, status_code=code) + + with pytest.raises(DownloadError): + wxee.download._download_url( + TEST_URL, TEST_OUT_DIR, progress=False, max_attempts=1 + ) + + +def test_download_url_fails_with_timeout(): + """Test that the download_url function fails correctly with a Timeout exception.""" + with requests_mock.Mocker() as m: + m.get(TEST_URL, exc=requests.exceptions.Timeout) + + with pytest.raises(DownloadError): + wxee.download._download_url( + TEST_URL, TEST_OUT_DIR, progress=False, max_attempts=1 + ) + + +def test_download_url_fails_with_connection_error(): + """Test that the download_url function fails correctly with a ConnectionError exception.""" + with requests_mock.Mocker() as m: + m.get(TEST_URL, exc=requests.exceptions.ConnectionError) + + with pytest.raises(DownloadError): + wxee.download._download_url( + TEST_URL, TEST_OUT_DIR, progress=False, max_attempts=1 + ) diff --git a/test/test_utils.py b/test/test_utils.py index 74301e9..81c3e08 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -7,8 +7,6 @@ import ee import pytest import rasterio -import requests -import requests_mock import wxee.utils @@ -198,35 +196,6 @@ def test_set_nodata(): os.remove(tmp_copy) -def test_download_url_creates_file(): - """Test that the download_url function downloads a mock file with correct content.""" - test_url = "http://aurl.com" - content = "this is the content of the file" - out_dir = os.path.join("test", "test_data") - - with requests_mock.Mocker() as m: - m.get(test_url, text=content) - file = wxee.utils._download_url(test_url, out_dir, False, 1) - - assert os.path.isfile(file) - - with open(file, "r") as result: - assert result.read() == content - - os.remove(file) - - -def test_download_url_fails_with_404(): - """Test that the download_url function fails correctly with a 404 response.""" - test_url = "http://aurl.com" - - with requests_mock.Mocker() as m: - m.get(test_url, text="", status_code=404) - - with pytest.raises(requests.exceptions.HTTPError): - wxee.utils._download_url(test_url, "", False, 1) - - def test_unpack_zip(): """Test that files can be correctly unpacked from a zip with matching file names.""" zip_path = os.path.join("test", "test_data", "test.zip")