319 lines
11 KiB
Python
319 lines
11 KiB
Python
# SPDX-FileCopyrightText: 2025 Blender Authors
|
|
#
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
|
|
"""
|
|
blender -b --factory-startup -P tests/python/bl_http_downloader.py -- --verbose
|
|
"""
|
|
|
|
|
|
__all__ = (
|
|
"main",
|
|
)
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from _bpy_internal.http.downloader import RequestDescription
|
|
|
|
|
|
output_dir: Path
|
|
|
|
|
|
class BasicImportTest(unittest.TestCase):
|
|
"""Just do a basic import and instantiation of classes.
|
|
|
|
This doesn't test the functionality, but does ensure that dependencies like
|
|
third-party libraries are available.
|
|
"""
|
|
|
|
def test_downloader(self) -> None:
|
|
from _bpy_internal.http import downloader as http_dl
|
|
|
|
metadata_provider = http_dl.MetadataProviderFilesystem(
|
|
cache_location=output_dir / "http_metadata"
|
|
)
|
|
downloader = http_dl.ConditionalDownloader(metadata_provider=metadata_provider)
|
|
self.assertIsNotNone(downloader)
|
|
|
|
def test_background_downloader(self) -> None:
|
|
from _bpy_internal.http import downloader as http_dl
|
|
|
|
metadata_provider = http_dl.MetadataProviderFilesystem(
|
|
cache_location=output_dir / "http_metadata"
|
|
)
|
|
options = http_dl.DownloaderOptions(
|
|
metadata_provider=metadata_provider,
|
|
timeout=1,
|
|
http_headers={'X-Unit-Test': self.__class__.__name__}
|
|
)
|
|
|
|
def on_error(req_desc: http_dl.RequestDescription, local_path: Path, ex: Exception) -> None:
|
|
self.fail(f"unexpected call to on_error({req_desc}, {local_path}, {ex})")
|
|
|
|
downloader = http_dl.BackgroundDownloader(
|
|
options=options,
|
|
on_callback_error=on_error,
|
|
)
|
|
|
|
# Test some trivial properties that don't require anything running.
|
|
self.assertTrue(downloader.all_downloads_done)
|
|
self.assertEqual(0, downloader.num_pending_downloads)
|
|
self.assertFalse(downloader.is_shutdown_requested)
|
|
self.assertFalse(downloader.is_shutdown_complete)
|
|
self.assertFalse(downloader.is_subprocess_alive)
|
|
|
|
|
|
class RequestDescriptionTest(unittest.TestCase):
|
|
def test_comparison(self) -> None:
|
|
from _bpy_internal.http import downloader as http_dl
|
|
|
|
url = "http://example.com/download.me"
|
|
http_descr = http_dl.RequestDescription("GET", url)
|
|
|
|
# Equality.
|
|
self.assertEqual(http_descr, http_descr, "Same instance")
|
|
self.assertEqual(
|
|
http_descr,
|
|
http_dl.RequestDescription("GET", url),
|
|
"Same HTTP method & URL, different instance",
|
|
)
|
|
self.assertEqual(
|
|
http_descr,
|
|
http_dl.RequestDescription("GET", url, response_headers={"a": "b"}),
|
|
"Same HTTP method & URL, different response headers",
|
|
)
|
|
|
|
# Inequality.
|
|
self.assertNotEqual(
|
|
http_descr,
|
|
http_dl.RequestDescription("DELETE", url),
|
|
"Different HTTP method, same URL",
|
|
)
|
|
self.assertNotEqual(
|
|
http_descr,
|
|
http_dl.RequestDescription("GET", "http://example.com/another-file.md"),
|
|
"Same HTTP method, different URL",
|
|
)
|
|
|
|
# Safety with other types.
|
|
self.assertNotEqual(http_descr, None, "comparison should be None-safe")
|
|
self.assertNotEqual(http_descr, "some other type", "comparison should be safe for other types")
|
|
|
|
def test_hash(self) -> None:
|
|
from _bpy_internal.http import downloader as http_dl
|
|
|
|
url = "http://example.com/download.me"
|
|
http_descr = http_dl.RequestDescription("GET", url)
|
|
|
|
# Equality.
|
|
self.assertEqual(hash(http_descr), hash(http_descr), "Same instance")
|
|
self.assertEqual(
|
|
hash(http_descr),
|
|
hash(http_dl.RequestDescription("GET", url)),
|
|
"Same HTTP method & URL, different instance",
|
|
)
|
|
self.assertEqual(
|
|
hash(http_descr),
|
|
hash(http_dl.RequestDescription("GET", url, response_headers={"a": "b"})),
|
|
"Same HTTP method & URL, different response headers",
|
|
)
|
|
|
|
# Inequality.
|
|
self.assertNotEqual(
|
|
hash(http_descr),
|
|
hash(http_dl.RequestDescription("DELETE", url)),
|
|
"Different HTTP method, same URL",
|
|
)
|
|
self.assertNotEqual(
|
|
hash(http_descr),
|
|
hash(http_dl.RequestDescription("GET", "http://example.com/another-file.md")),
|
|
"Same HTTP method, different URL",
|
|
)
|
|
|
|
# Safety with other types.
|
|
self.assertNotEqual(hash(http_descr), hash(None))
|
|
self.assertNotEqual(hash(RequestDescription("", "")), hash(None))
|
|
|
|
|
|
class MaxDownloadSizeTest(unittest.TestCase):
|
|
def test_max_size(self) -> None:
|
|
from _bpy_internal.http import downloader as http_dl
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
metadata_provider = MagicMock(spec=http_dl.MetadataProvider)
|
|
metadata_provider.load.return_value = None
|
|
|
|
downloader = http_dl.ConditionalDownloader(metadata_provider=metadata_provider)
|
|
downloader.max_disk_size_bytes = 100
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.headers = {"Content-Length": "200", "Content-Type": "text/plain"}
|
|
mock_response.status_code = 200
|
|
# When used as context manager, return itself.
|
|
mock_response.__enter__.return_value = mock_response
|
|
|
|
# NOTE: when this test fails, it will say "TypeError: a bytes-like object is required, not 'MagicMock'". This
|
|
# indicates that the HTTP downloader tried to actually read bytes from the mock response, which it doesn't
|
|
# support.
|
|
|
|
with patch.object(http_dl.requests.Session, 'send', return_value=mock_response):
|
|
with self.assertRaises(http_dl.ContentLengthTooBigError):
|
|
downloader.download_to_file("https://example.com/huge.json", Path("/tmp/huge.json"))
|
|
|
|
|
|
class BackgroundDownloaderProcessTest(unittest.TestCase):
|
|
"""Start & stop the background process for the BackgroundDownloader.
|
|
|
|
This doesn't test any HTTP requests, but does start & stop the background
|
|
process to check that this is at least possible.
|
|
"""
|
|
|
|
def test_start_stop(self) -> None:
|
|
from _bpy_internal.http import downloader as http_dl
|
|
|
|
metadata_provider = http_dl.MetadataProviderFilesystem(
|
|
cache_location=output_dir / "http_metadata"
|
|
)
|
|
options = http_dl.DownloaderOptions(
|
|
metadata_provider=metadata_provider,
|
|
timeout=1,
|
|
http_headers={'X-Unit-Test': self.__class__.__name__}
|
|
)
|
|
|
|
def on_error(req_desc: http_dl.RequestDescription, local_path: Path, ex: Exception) -> None:
|
|
self.fail(f"unexpected call to on_error({req_desc}, {local_path}, {ex})")
|
|
|
|
downloader = http_dl.BackgroundDownloader(
|
|
options=options,
|
|
on_callback_error=on_error,
|
|
)
|
|
|
|
# Queueing a download before the downloader has started should be rejected.
|
|
with self.assertRaises(RuntimeError):
|
|
downloader.queue_download("https://example.com/", output_dir / "download.tmp")
|
|
|
|
downloader.start()
|
|
|
|
try:
|
|
self.assertFalse(downloader.is_shutdown_requested)
|
|
self.assertFalse(downloader.is_shutdown_complete)
|
|
self.assertTrue(downloader.is_subprocess_alive)
|
|
|
|
# For good measure, call the update function a few times to ensure that
|
|
# any messages are sent. There shouldn't be any, but this should also
|
|
# not be a problem.
|
|
downloader.update()
|
|
downloader.update()
|
|
downloader.update()
|
|
finally:
|
|
# In case any of the pre-shutdown assertions fail, the downloader
|
|
# should still be shut down.
|
|
if downloader.is_subprocess_alive:
|
|
downloader.shutdown()
|
|
|
|
downloader.shutdown()
|
|
|
|
self.assertTrue(downloader.is_shutdown_requested)
|
|
self.assertTrue(downloader.is_shutdown_complete)
|
|
self.assertFalse(downloader.is_subprocess_alive)
|
|
|
|
|
|
class ConditionalDownloaderTest(unittest.TestCase):
|
|
|
|
def test_request_cancel(self) -> None:
|
|
"""Request and immediately cancel a download.
|
|
|
|
This doesn't test actual HTTP traffic, but rather cancels the request
|
|
at the first possible moment.
|
|
"""
|
|
from _bpy_internal.http import downloader as http_dl
|
|
|
|
# Create the downloader.
|
|
metadata_provider = http_dl.MetadataProviderFilesystem(
|
|
cache_location=output_dir / "http_metadata"
|
|
)
|
|
downloader = http_dl.ConditionalDownloader(metadata_provider=metadata_provider)
|
|
|
|
download_url = "http://localhost:47/download.me"
|
|
download_path = output_dir / "the_download"
|
|
|
|
# Set up the periodic check callback function so it immediately cancels.
|
|
expected_http_descr = http_dl.RequestDescription("GET", download_url)
|
|
periodic_check_called = False
|
|
|
|
def periodic_check(http_req_descr: http_dl.RequestDescription) -> bool:
|
|
nonlocal periodic_check_called
|
|
|
|
periodic_check_called = True
|
|
self.assertEqual(expected_http_descr, http_req_descr)
|
|
|
|
return False
|
|
downloader.periodic_check = periodic_check
|
|
|
|
# Set up a reporter.
|
|
reporter = http_dl.QueueingReporter()
|
|
downloader.add_reporter(reporter)
|
|
|
|
# Request the download.
|
|
with self.assertRaises(http_dl.DownloadCancelled) as cancel_ex:
|
|
_ = downloader.download_to_file(download_url, download_path)
|
|
|
|
# Check the result.
|
|
self.assertFalse(download_path.exists())
|
|
self.assertTrue(periodic_check_called)
|
|
self.assertEqual(
|
|
[
|
|
('download_starts', (expected_http_descr, )),
|
|
('download_error', (expected_http_descr, download_path, cancel_ex.exception)),
|
|
],
|
|
list(reporter._queue) # pyright: ignore[reportPrivateUsage]
|
|
)
|
|
|
|
|
|
class HumanizeSizeTest(unittest.TestCase):
|
|
def test_humanize_size(self):
|
|
from _bpy_internal.http.downloader import humanize_size
|
|
|
|
self.assertEqual(humanize_size(0), "0 B")
|
|
self.assertEqual(humanize_size(1), "1 B")
|
|
self.assertEqual(humanize_size(1023), "1023 B")
|
|
|
|
self.assertEqual(humanize_size(1024), "1 KiB")
|
|
self.assertEqual(humanize_size(1024**2), "1 MiB")
|
|
self.assertEqual(humanize_size(1024**3), "1 GiB")
|
|
self.assertEqual(humanize_size(1024**4), "1 TiB")
|
|
|
|
self.assertEqual(humanize_size(1536), "1.5 KiB")
|
|
self.assertEqual(humanize_size(5046586573), "4.7 GiB")
|
|
|
|
# 47.0... GiB -> trailing zero stripped
|
|
self.assertEqual(humanize_size(50465865728), "47 GiB")
|
|
|
|
# Beyond the last unit should also be ok.
|
|
self.assertEqual(humanize_size(1024**7), "1024 EiB")
|
|
self.assertEqual(humanize_size(1024**8), "1048576 EiB")
|
|
|
|
# Negative values should be handled like positive values.
|
|
self.assertEqual(humanize_size(-1), "-1 B")
|
|
self.assertEqual(humanize_size(-2 * 1024**2), "-2 MiB")
|
|
|
|
|
|
def main() -> None:
|
|
global output_dir
|
|
|
|
import sys
|
|
import tempfile
|
|
|
|
argv = [sys.argv[0]]
|
|
if '--' in sys.argv:
|
|
argv.extend(sys.argv[sys.argv.index('--') + 1:])
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_dir = Path(temp_dir)
|
|
unittest.main(argv=argv)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|