diff --git a/docs/howto/use-fetches.rst b/docs/howto/use-fetches.rst index d7e727b83..faa240f97 100644 --- a/docs/howto/use-fetches.rst +++ b/docs/howto/use-fetches.rst @@ -105,3 +105,35 @@ There are a few differences from the earlier ``build`` examples here: It is not possible to configure the ``dest`` or ``extract`` values when using ``fetch`` or ``toolchain`` kinds. + +Tuning Download Performance +--------------------------- + +``fetch-content`` already downloads the artifacts of a task concurrently, but +each one is fetched over a single connection, which can leave a fast worker's +network idle when a task depends on one or two large artifacts. The following +environment variables tune this. Both are off by default, because whether +either helps depends on where a worker sits relative to the artifact storage; +measure with the ``fetch_content`` Perfherder suite before turning them on for +a worker pool. + +``TASKGRAPH_FETCH_SLICES`` + Number of concurrent HTTP range requests to split a single download into. + Defaults to ``1``, which disables slicing. Servers that don't support range + requests fall back to a single stream automatically. + +``TASKGRAPH_FETCH_SLICE_MIN_BYTES`` + Downloads smaller than this are never sliced. Defaults to 64MB. + +``TASKGRAPH_SKIP_CDN`` + Set to ``1`` to send the ``x-taskcluster-skip-cdn`` header when fetching + task artifacts, which asks the queue to redirect to the bucket backing the + artifact rather than to the CDN in front of it. The queue does this + automatically when it recognises the caller as running in the same region + as the bucket, but it can only recognise EC2 instances, so workers running + anywhere else have to opt in. + + This is a trade off rather than a clear win. Going direct can be quicker + for a worker close to the bucket, but it gives up the CDN cache that every + other worker is sharing, and it is markedly slower for a worker that isn't + nearby. diff --git a/src/taskgraph/run-task/fetch-content b/src/taskgraph/run-task/fetch-content index 94bb2835d..027991934 100755 --- a/src/taskgraph/run-task/fetch-content +++ b/src/taskgraph/run-task/fetch-content @@ -24,6 +24,7 @@ import subprocess import sys import tarfile import tempfile +import threading import time import urllib.parse import urllib.request @@ -42,12 +43,52 @@ except ImportError: CONCURRENCY = multiprocessing.cpu_count() +# Number of concurrent HTTP range requests to split a single download into. +# 1 disables slicing entirely, which is the default: whether slicing helps +# depends on how much bandwidth a single connection can get on a given worker, +# so it is opt-in via TASKGRAPH_FETCH_SLICES until measured. +DEFAULT_SLICE_COUNT = 1 + +# Downloads smaller than this are never sliced. The extra round trip needed to +# discover the size costs more than the parallelism saves. +DEFAULT_SLICE_MIN_BYTES = 64 * 1024 * 1024 + +CHUNK_SIZE = 65536 + +# How much the request that probes for range support downloads. Large enough +# that the round trip isn't wasted on small files -- most are fetched outright +# -- and small enough not to delay the concurrent slices by much. +PROBE_BYTES = 1024 * 1024 + +# Slices below this size don't earn a connection of their own. +MIN_SLICE_BYTES = 1024 * 1024 + +_slice_pool = None +_slice_pool_lock = threading.Lock() + def log(msg): print(msg, file=sys.stderr) sys.stderr.flush() +def env_flag(name): + """Whether the named environment variable is set to a truthy value.""" + return os.environ.get(name, "").strip().lower() in ("1", "true", "yes") + + +def env_int(name, default): + """Value of the named environment variable as an int, or ``default``.""" + value = os.environ.get(name, "").strip() + if not value: + return default + try: + return int(value) + except ValueError: + log(f"Ignoring non-integer value for {name}: {value}") + return default + + class IntegrityError(Exception): """Represents an integrity error when downloading a URL.""" @@ -161,6 +202,249 @@ def retrier(attempts=5, sleeptime=10, max_sleeptime=300, sleepscale=1.5, jitter= time.sleep(sleeptime_real) +def parse_headers(headers): + """Turn a list of ``"Key: value"`` strings into a dict.""" + parsed = {} + for header in headers or []: + key, val = header.split(":", 1) + parsed[key.strip()] = val.strip() + return parsed + + +def urlopen(req, timeout=60): + """``urllib.request.urlopen``, using certifi's CA bundle when available.""" + kwargs = {} + if certifi: + kwargs["context"] = ssl.create_default_context(cafile=certifi.where()) + return urllib.request.urlopen(req, timeout=timeout, **kwargs) + + +def get_slice_pool(): + """Thread pool shared by all sliced downloads. + + ``fetch_urls`` already runs one thread per download, and each of those may + ask for several slices. Funnelling the slices through a single pool keeps + the number of in-flight connections bounded by CONCURRENCY rather than + multiplying the two together. + """ + global _slice_pool + with _slice_pool_lock: + if _slice_pool is None: + _slice_pool = concurrent.futures.ThreadPoolExecutor( + CONCURRENCY, thread_name_prefix="fetch-slice" + ) + return _slice_pool + + +def range_headers(headers, start, end): + parsed = dict(headers) + # Ranges and transfer encodings don't mix: we want byte offsets into the + # stored object, not into a gzip stream. + parsed["Accept-Encoding"] = "identity" + parsed["Range"] = f"bytes={start}-{end}" + return parsed + + +def write_slice(src, path, start, end): + """Copy ``src`` into ``path`` at ``start``, checking it stops at ``end``.""" + expected = end - start + 1 + written = 0 + with path.open("r+b") as dst: + dst.seek(start) + while True: + chunk = src.read(CHUNK_SIZE) + if not chunk: + break + written += len(chunk) + if written > expected: + raise IntegrityError( + f"bytes={start}-{end} returned more than the {expected} " + "bytes requested" + ) + dst.write(chunk) + + if written != expected: + raise IntegrityError( + f"bytes={start}-{end}: wanted {expected} bytes; got {written}" + ) + + +def download_first_slice(url, path, length, headers): + """Download the first ``length`` bytes of ``url`` to the start of ``path``. + + Doubles as a probe for range support. Returns a ``(url, end, size)`` tuple, + where the URL is the one the request finally resolved to -- so the + remaining slices don't each have to redo the redirects -- ``end`` is the + last offset written, and ``size`` is the total length of the content. They + are all ``None``, and nothing is written, if the server didn't honour the + range request. + """ + req = urllib.request.Request(url, None, range_headers(headers, 0, length - 1)) + try: + with urlopen(req) as src: + if src.getcode() != 206: + return None, None, None + + match = re.match( + r"^bytes 0-(\d+)/(\d+)$", + (src.getheader("content-range") or "").strip(), + ) + if not match: + return None, None, None + + end, total = int(match.group(1)), int(match.group(2)) + final_url = src.geturl() + + # The file doesn't exist yet and its final size wasn't known until + # now, so create it before writing into it. It gets extended to the + # full size once the caller knows there is more to come. + with path.open("wb"): + pass + write_slice(src, path, 0, end) + except urllib.error.HTTPError as e: + # A zero length object makes any range unsatisfiable, so an empty + # artifact lands here rather than returning a 206 covering the whole + # (empty) object. Any other error will resurface on the single stream + # fallback, where it gets reported the same way it always has been. + log(f"Range request for {url} returned {e.code}; not slicing") + return None, None, None + + return final_url, end, total + + +def sha256_path(path): + h = hashlib.sha256() + with path.open("rb") as fh: + while True: + chunk = fh.read(CHUNK_SIZE) + if not chunk: + break + h.update(chunk) + return h.hexdigest() + + +def download_slice(url, path, start, end, headers, attempts=3): + """Download ``bytes=start-end`` of ``url`` into ``path`` at that offset.""" + for attempt in range(attempts): + try: + req = urllib.request.Request(url, None, range_headers(headers, start, end)) + with urlopen(req) as src: + if src.getcode() != 206: + raise IntegrityError( + f"expected a 206 response for bytes={start}-{end} of " + f"{url}; got {src.getcode()}" + ) + write_slice(src, path, start, end) + return + except Exception as e: + log(f"Slice bytes={start}-{end} of {url} failed: {e}") + if attempt == attempts - 1: + raise + time.sleep(2**attempt) + + +def remaining_slices(start, total, slices): + """Split ``start``..``total`` into ``slices`` inclusive ranges.""" + remaining = total - start + if remaining <= 0: + return [] + + slices = max(1, min(slices, remaining // MIN_SLICE_BYTES)) + length = remaining // slices + return [ + ( + start + i * length, + total - 1 if i == slices - 1 else start + (i + 1) * length - 1, + ) + for i in range(slices) + ] + + +def sliced_download_to_path(url, path, sha256=None, size=None, headers=None): + """Download ``url`` to ``path`` using concurrent HTTP range requests. + + Returns True if the download happened, and False if slicing is disabled or + the server doesn't support range requests, in which case the caller should + fall back to a single stream. Raises if the download itself fails. + """ + slices = env_int("TASKGRAPH_FETCH_SLICES", DEFAULT_SLICE_COUNT) + min_bytes = env_int("TASKGRAPH_FETCH_SLICE_MIN_BYTES", DEFAULT_SLICE_MIN_BYTES) + if slices < 2: + return False + + # When the size is known up front, small downloads can be ruled out without + # issuing any requests at all. + if size is not None and size < min_bytes: + return False + + parsed_headers = parse_headers(headers) + tmp = path.with_name(f"{path.name}.tmp") + t0 = time.time() + + try: + final_url, probe_end, total = download_first_slice( + url, tmp, PROBE_BYTES, parsed_headers + ) + if total is None: + log(f"Cannot slice {url}; downloading it in a single stream") + return False + + if size is not None and size != total: + raise IntegrityError( + f"size mismatch on {url}: wanted {size}; range request reports {total}" + ) + + with tmp.open("r+b") as fh: + fh.truncate(total) + + # The size wasn't known until the probe came back. Now that the first + # chunk is on disk there is no point falling back, so finish + # under-sized downloads with a single request for the remainder. + ranges = remaining_slices( + probe_end + 1, total, slices if total >= min_bytes else 1 + ) + if ranges: + log(f"Downloading {url} ({total} bytes) in {len(ranges) + 1} slices") + pool = get_slice_pool() + futures = [ + pool.submit(download_slice, final_url, tmp, start, end, parsed_headers) + for start, end in ranges + ] + + error = None + for future in futures: + try: + future.result() + except Exception as e: + if error is None: + error = e + # Don't start slices that haven't been picked up yet, + # but keep waiting on the rest so that nothing is still + # writing to the file when we remove it. + for pending in futures: + pending.cancel() + if error is not None: + raise error + + if sha256: + digest = sha256_path(tmp) + if digest != sha256: + raise IntegrityError( + f"sha256 mismatch on {url}: wanted {sha256}; got {digest}" + ) + log(f"Verified sha256 integrity of {url}") + except Exception: + try: + tmp.unlink() + except FileNotFoundError: + pass + raise + + log(f"{url} resolved to {total} bytes in {time.time() - t0:.3f}s") + tmp.rename(path) + return True + + def stream_download(url, sha256=None, size=None, headers=None): """Download a URL to a generator, optionally with content verification. @@ -182,16 +466,10 @@ def stream_download(url, sha256=None, size=None, headers=None): t0 = time.time() req_headers = {"Accept-Encoding": "gzip"} - for header in headers: - key, val = header.split(":") - req_headers[key.strip()] = val.strip() + req_headers.update(parse_headers(headers)) req = urllib.request.Request(url, None, req_headers) - kwargs = {} - if certifi: - ssl_context = ssl.create_default_context(cafile=certifi.where()) - kwargs["context"] = ssl_context - with urllib.request.urlopen(req, timeout=60, **kwargs) as fh: + with urlopen(req) as fh: if not url.endswith(".gz") and fh.info().get("Content-Encoding") == "gzip": fh = gzip.GzipFile(fileobj=fh) else: @@ -262,6 +540,11 @@ def download_to_path(url, path, sha256=None, size=None, headers=None): try: log(f"Downloading {url} to {path}") + if sliced_download_to_path( + url, path, sha256=sha256, size=size, headers=headers + ): + return + with rename_after_close(path, "wb") as fh: for chunk in stream_download( url, sha256=sha256, size=size, headers=headers @@ -618,7 +901,9 @@ def repack_archive( ) -def fetch_and_extract(url, dest_dir, extract=True, sha256=None, size=None): +def fetch_and_extract( + url, dest_dir, extract=True, sha256=None, size=None, headers=None +): """Fetch a URL and extract it to a destination path. If the downloaded URL is an archive, it is extracted automatically @@ -629,7 +914,7 @@ def fetch_and_extract(url, dest_dir, extract=True, sha256=None, size=None): basename = urllib.parse.unquote(urllib.parse.urlparse(url).path.split("/")[-1]) dest_path = dest_dir / basename - download_to_path(url, dest_path, sha256=sha256, size=size) + download_to_path(url, dest_path, sha256=sha256, size=size, headers=headers) if not extract: return @@ -909,6 +1194,18 @@ def get_hash(fetch, root_url): def command_task_artifacts(args): start = time.monotonic() fetches = json.loads(os.environ["MOZ_FETCHES"]) + + # Ask the queue to redirect us straight at the bucket backing the artifact + # rather than at the CDN in front of it. The queue does this automatically + # when it recognises the caller as running in the same region as the + # bucket, but it can only recognise EC2 instances, so workers running + # anywhere else have to opt in. It is a trade off rather than a clear win: + # it can be quicker for a worker that is close to the bucket, at the cost + # of missing the cache that every other worker is sharing. + headers = ( + ["x-taskcluster-skip-cdn: true"] if env_flag("TASKGRAPH_SKIP_CDN") else None + ) + downloads = [] for fetch in fetches: extdir = pathlib.Path(args.dest) @@ -932,7 +1229,7 @@ def command_task_artifacts(args): task=fetch["task"], artifact=encoded_artifact, ) - downloads.append((url, extdir, fetch["extract"], sha256)) + downloads.append((url, extdir, fetch["extract"], sha256, None, headers)) fetch_urls(downloads) end = time.monotonic() diff --git a/test/test_scripts_fetch_content.py b/test/test_scripts_fetch_content.py index 658e0c8c2..5d681876c 100644 --- a/test/test_scripts_fetch_content.py +++ b/test/test_scripts_fetch_content.py @@ -1,6 +1,9 @@ +import hashlib +import http.server import json import os import pathlib +import threading import urllib.request from importlib.machinery import SourceFileLoader from importlib.util import module_from_spec, spec_from_loader @@ -29,6 +32,308 @@ def fetch_content_mod(): return mod +class RangeServer(http.server.ThreadingHTTPServer): + """Serves a single blob, optionally honouring range requests.""" + + daemon_threads = True + allow_reuse_address = True + + def __init__(self, content, ranges=True): + super().__init__(("127.0.0.1", 0), RangeHandler) + self.content = content + self.ranges = ranges + self.requests = [] + self.lock = threading.Lock() + + @property + def url(self): + return "http://{}:{}/blob".format(*self.server_address) + + +class RangeHandler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_GET(self): + content = self.server.content + requested = self.headers.get("Range") + with self.server.lock: + self.server.requests.append(requested) + + start, end = 0, len(content) - 1 + partial = False + if requested and self.server.ranges: + start, _, last = requested.partition("=")[2].partition("-") + start, end = int(start), int(last) + if start >= len(content): + # An empty object makes every range unsatisfiable. + self.send_response(416) + self.send_header("Content-Range", f"bytes */{len(content)}") + self.send_header("Content-Length", "0") + self.end_headers() + return + end = min(end, len(content) - 1) + partial = True + + body = content[start : end + 1] + self.send_response(206 if partial else 200) + if partial: + self.send_header("Content-Range", f"bytes {start}-{end}/{len(content)}") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +@pytest.fixture +def serve(): + servers = [] + + def inner(content, ranges=True): + server = RangeServer(content, ranges=ranges) + threading.Thread( + target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True + ).start() + servers.append(server) + return server + + yield inner + + for server in servers: + server.shutdown() + server.server_close() + + +@pytest.fixture +def sliced(monkeypatch, fetch_content_mod): + """Enable slicing with boundaries small enough to exercise in a test.""" + + def inner(slices=4, min_bytes=1024, probe=256): + monkeypatch.setenv("TASKGRAPH_FETCH_SLICES", str(slices)) + monkeypatch.setenv("TASKGRAPH_FETCH_SLICE_MIN_BYTES", str(min_bytes)) + monkeypatch.setattr(fetch_content_mod, "PROBE_BYTES", probe) + # remaining_slices refuses to make slices smaller than a megabyte, + # which would collapse every test case down to a single slice. + monkeypatch.setattr(fetch_content_mod, "MIN_SLICE_BYTES", 64) + + return inner + + +def test_sliced_download(tmp_path, fetch_content_mod, serve, sliced): + content = os.urandom(4096) + server = serve(content) + sliced(slices=4) + dest = tmp_path / "blob" + + assert fetch_content_mod.sliced_download_to_path( + server.url, + dest, + sha256=hashlib.sha256(content).hexdigest(), + size=len(content), + ) + assert dest.read_bytes() == content + # The probe, plus one request per remaining slice. + assert len(server.requests) == 5 + assert server.requests[0] == "bytes=0-255" + + +def test_sliced_download_reassembles_out_of_order( + tmp_path, fetch_content_mod, serve, sliced +): + """Slices land at the right offsets no matter what order they finish in.""" + content = bytes(i % 251 for i in range(100000)) + server = serve(content) + sliced(slices=8) + dest = tmp_path / "blob" + + assert fetch_content_mod.sliced_download_to_path(server.url, dest) + assert dest.read_bytes() == content + + +def test_sliced_download_disabled(tmp_path, fetch_content_mod, serve, monkeypatch): + server = serve(b"x" * 4096) + monkeypatch.setenv("TASKGRAPH_FETCH_SLICES", "1") + dest = tmp_path / "blob" + + assert not fetch_content_mod.sliced_download_to_path(server.url, dest) + assert server.requests == [] + assert not dest.exists() + + +def test_sliced_download_known_small_size_skips_probe( + tmp_path, fetch_content_mod, serve, sliced +): + content = b"x" * 512 + server = serve(content) + sliced(slices=4, min_bytes=1024) + dest = tmp_path / "blob" + + assert not fetch_content_mod.sliced_download_to_path( + server.url, dest, size=len(content) + ) + assert server.requests == [] + + +def test_sliced_download_unknown_small_size_finishes_in_probe( + tmp_path, fetch_content_mod, serve, sliced +): + """A file the probe swallows whole costs exactly one request.""" + content = b"x" * 200 + server = serve(content) + sliced(slices=4, min_bytes=1024, probe=256) + dest = tmp_path / "blob" + + assert fetch_content_mod.sliced_download_to_path(server.url, dest) + assert dest.read_bytes() == content + assert len(server.requests) == 1 + + +def test_sliced_download_no_range_support(tmp_path, fetch_content_mod, serve, sliced): + server = serve(b"x" * 4096, ranges=False) + sliced() + dest = tmp_path / "blob" + + assert not fetch_content_mod.sliced_download_to_path(server.url, dest) + assert not dest.exists() + + +def test_sliced_download_empty_object(tmp_path, fetch_content_mod, serve, sliced): + """A zero length artifact 416s, and must fall back rather than blow up.""" + server = serve(b"") + sliced() + dest = tmp_path / "blob" + + assert not fetch_content_mod.sliced_download_to_path(server.url, dest) + assert not dest.exists() + + # ...and the fallback still downloads it. + fetch_content_mod.download_to_path(server.url, dest) + assert dest.read_bytes() == b"" + + +def test_sliced_download_bad_sha256(tmp_path, fetch_content_mod, serve, sliced): + server = serve(os.urandom(4096)) + sliced() + dest = tmp_path / "blob" + + with pytest.raises(fetch_content_mod.IntegrityError): + fetch_content_mod.sliced_download_to_path(server.url, dest, sha256="0" * 64) + + assert not dest.exists() + assert not dest.with_name(f"{dest.name}.tmp").exists() + + +def test_sliced_download_bad_size(tmp_path, fetch_content_mod, serve, sliced): + server = serve(os.urandom(4096)) + sliced() + dest = tmp_path / "blob" + + with pytest.raises(fetch_content_mod.IntegrityError): + fetch_content_mod.sliced_download_to_path(server.url, dest, size=9999) + + assert not dest.exists() + assert not dest.with_name(f"{dest.name}.tmp").exists() + + +def test_sliced_download_forwards_headers(tmp_path, fetch_content_mod, serve, sliced): + """Caller supplied headers reach every request, not just the probe.""" + seen = [] + + class Recording(RangeHandler): + def do_GET(self): + seen.append(self.headers.get("X-Taskcluster-Skip-Cdn")) + super().do_GET() + + server = serve(os.urandom(4096)) + server.RequestHandlerClass = Recording + sliced(slices=4) + + assert fetch_content_mod.sliced_download_to_path( + server.url, tmp_path / "blob", headers=["x-taskcluster-skip-cdn: true"] + ) + assert seen == ["true"] * 5 + + +@pytest.mark.parametrize( + "start,total,slices,expected", + ( + pytest.param( + 0, 400, 4, [(0, 99), (100, 199), (200, 299), (300, 399)], id="even" + ), + pytest.param( + 0, 10, 4, [(0, 1), (2, 3), (4, 5), (6, 9)], id="remainder to last" + ), + pytest.param(64, 128, 2, [(64, 95), (96, 127)], id="offset start"), + pytest.param(100, 100, 4, [], id="nothing left"), + pytest.param(200, 100, 4, [], id="past the end"), + pytest.param(0, 100, 1, [(0, 99)], id="single"), + ), +) +def test_remaining_slices( + fetch_content_mod, monkeypatch, start, total, slices, expected +): + monkeypatch.setattr(fetch_content_mod, "MIN_SLICE_BYTES", 1) + assert fetch_content_mod.remaining_slices(start, total, slices) == expected + if expected: + # The ranges must tile the region exactly, with no gaps or overlaps. + assert expected[0][0] == start + assert expected[-1][1] == total - 1 + for (_, prev_end), (next_start, _) in zip(expected, expected[1:]): + assert next_start == prev_end + 1 + + +def test_remaining_slices_respects_minimum(fetch_content_mod): + """Slices below the minimum are merged rather than each taking a connection.""" + assert fetch_content_mod.remaining_slices(0, 1024, 8) == [(0, 1023)] + + +@pytest.mark.parametrize( + "headers,expected", + ( + pytest.param(["Foo: bar"], {"Foo": "bar"}, id="simple"), + pytest.param([], {}, id="empty"), + pytest.param(None, {}, id="none"), + pytest.param( + ["Location: https://example.com:443/x"], + {"Location": "https://example.com:443/x"}, + id="colon in value", + ), + ), +) +def test_parse_headers(fetch_content_mod, headers, expected): + assert fetch_content_mod.parse_headers(headers) == expected + + +@pytest.mark.parametrize( + "value,expected", + ( + pytest.param(None, None, id="unset"), + pytest.param("1", ["x-taskcluster-skip-cdn: true"], id="1"), + pytest.param("true", ["x-taskcluster-skip-cdn: true"], id="true"), + pytest.param("0", None, id="0"), + ), +) +def test_command_task_artifacts_skip_cdn( + monkeypatch, tmp_path, fetch_content_mod, value, expected +): + fetches = [{"task": "abc123", "artifact": "public/foo.zip", "extract": False}] + monkeypatch.setenv("MOZ_FETCHES", json.dumps(fetches)) + monkeypatch.setenv("TASKCLUSTER_ROOT_URL", "https://tc.example.com") + monkeypatch.delenv("TASKGRAPH_SKIP_CDN", raising=False) + if value is not None: + monkeypatch.setenv("TASKGRAPH_SKIP_CDN", value) + + captured = [] + monkeypatch.setattr(fetch_content_mod, "fetch_urls", captured.extend) + + args = MagicMock() + args.dest = str(tmp_path) + fetch_content_mod.command_task_artifacts(args) + + assert [download[5] for download in captured] == [expected] + + @pytest.mark.parametrize( "url,sha256,size,headers,raises", ( @@ -128,7 +433,7 @@ def test_command_task_artifacts_url_encoding( captured_urls = [] def mock_fetch_urls(downloads): - for url, dest_dir, extract, sha256 in downloads: + for url, dest_dir, extract, sha256, size, headers in downloads: captured_urls.append(url) monkeypatch.setattr(fetch_content_mod, "fetch_urls", mock_fetch_urls) @@ -166,7 +471,7 @@ def test_fetch_and_extract_dest_filename( ): downloaded_to = [] - def mock_download_to_path(url, path, sha256=None, size=None): + def mock_download_to_path(url, path, sha256=None, size=None, headers=None): downloaded_to.append(path) path.touch()