diff --git a/tests/core/test_adaptive_workers.py b/tests/core/test_adaptive_workers.py
index 1f45aad6..fd3fdd2c 100644
--- a/tests/core/test_adaptive_workers.py
+++ b/tests/core/test_adaptive_workers.py
@@ -30,7 +30,7 @@ WINDOW = 10.0
DT = 0.5
-def _fill_window(c: AdaptiveWorkerController, rate: float, now: float) -> float:
+def fill_window(c: AdaptiveWorkerController, rate: float, now: float) -> float:
"""Record ``WINDOW`` seconds of samples at ``rate`` bytes/sec ending on the 0.5s grid.
Records a sample at ``now`` and every 0.5s up to ``now + WINDOW`` so the oldest kept
@@ -44,7 +44,7 @@ def _fill_window(c: AdaptiveWorkerController, rate: float, now: float) -> float:
return now
-def _run_tick(c: AdaptiveWorkerController, rate: float, now: float, errors: int = 0) -> tuple[int, float]:
+def run_tick(c: AdaptiveWorkerController, rate: float, now: float, errors: int = 0) -> tuple[int, float]:
"""Advance one tick recording samples at ``rate``, optionally inject ``errors``, then update."""
end = now + TICK
while now < end - 1e-9:
@@ -78,11 +78,11 @@ def test_ramps_up_while_speed_improves() -> None:
now = 0.0
c.update(now) # arm
rate = 1_000_000.0
- now = _fill_window(c, rate, now)
+ now = fill_window(c, rate, now)
targets = [c.update(now)] # first evaluation: slow-start doubles 6 -> 12
for _ in range(6):
rate *= 2 # sustained >10% windowed gain each tick
- target, now = _run_tick(c, rate, now)
+ target, now = run_tick(c, rate, now)
targets.append(target)
assert targets == sorted(targets) # monotonic non-decreasing
assert targets[-1] == 16 # climbed to the cap
@@ -94,13 +94,13 @@ def test_reverts_on_plateau_and_ends_slow_start() -> None:
now = 0.0
c.update(now)
rate = 1_000_000.0
- now = _fill_window(c, rate, now)
+ now = fill_window(c, rate, now)
assert c.update(now) == 12 # first increase slow-starts 6 -> 12
# hold the rate flat: <10% gain -> the tick reverts the increase to its restore point
- target, now = _run_tick(c, rate, now)
+ target, now = run_tick(c, rate, now)
assert target == 6
# slow-start is over: the next successful probe is additive (+2), not a double
- target, now = _run_tick(c, rate * 4, now)
+ target, now = run_tick(c, rate * 4, now)
assert target == 8
@@ -108,17 +108,17 @@ def test_halves_and_cooldown_on_error_burst() -> None:
c = AdaptiveWorkerController(cap=16, tick=TICK, window=WINDOW)
now = 0.0
c.update(now)
- now = _fill_window(c, 1_000_000.0, now)
+ now = fill_window(c, 1_000_000.0, now)
for _ in range(3):
c.record_error(now)
assert c.update(now) == 3 # 3 errors -> multiplicative decrease 6 // 2
# cooldown tick holds even with improving throughput
- target, now = _run_tick(c, 2_000_000.0, now)
+ target, now = run_tick(c, 2_000_000.0, now)
assert target == 3
# cooldown consumed -> climbing resumes
- target, now = _run_tick(c, 4_000_000.0, now)
+ target, now = run_tick(c, 4_000_000.0, now)
assert target == 5
@@ -128,11 +128,11 @@ def test_clamps_to_bounds_and_never_exceeds_cap() -> None:
now = 0.0
c.update(now)
rate = 1_000_000.0
- now = _fill_window(c, rate, now)
+ now = fill_window(c, rate, now)
seen = [c.update(now)]
for _ in range(10):
rate *= 2
- target, now = _run_tick(c, rate, now)
+ target, now = run_tick(c, rate, now)
seen.append(target)
assert all(2 <= s <= 4 for s in seen)
assert max(seen) == 4 # never exceeds the cap
@@ -142,7 +142,7 @@ def test_error_burst_floors_at_min() -> None:
c = AdaptiveWorkerController(cap=16, start=2, tick=TICK, window=WINDOW)
now = 0.0
c.update(now)
- now = _fill_window(c, 1_000_000.0, now)
+ now = fill_window(c, 1_000_000.0, now)
for _ in range(5):
c.record_error(now)
assert c.update(now) == 2 # max(ADAPTIVE_MIN, 2 // 2)
@@ -153,7 +153,7 @@ def test_tail_guard_suppresses_increase() -> None:
c = AdaptiveWorkerController(cap=16, tick=TICK, window=WINDOW)
now = 0.0
c.update(now) # arm at target 6
- now = _fill_window(c, 1_000_000.0, now)
+ now = fill_window(c, 1_000_000.0, now)
assert c.update(now, 3) == 6 # 3 < target 6 -> holds at 6 rather than climbing to 8
@@ -162,7 +162,7 @@ def test_tail_guard_none_path_unchanged() -> None:
c = AdaptiveWorkerController(cap=16, tick=TICK, window=WINDOW)
now = 0.0
c.update(now)
- now = _fill_window(c, 1_000_000.0, now)
+ now = fill_window(c, 1_000_000.0, now)
assert c.update(now) == 12 # first evaluation slow-starts 6 -> 12
@@ -171,7 +171,7 @@ def test_tail_guard_still_halves_on_error_burst() -> None:
c = AdaptiveWorkerController(cap=16, tick=TICK, window=WINDOW)
now = 0.0
c.update(now)
- now = _fill_window(c, 1_000_000.0, now)
+ now = fill_window(c, 1_000_000.0, now)
for _ in range(3):
c.record_error(now)
assert c.update(now, 2) == 3 # guard active (2 < 6) but 3 errors -> 6 // 2
diff --git a/tests/core/test_dash_collapse_integration.py b/tests/core/test_dash_collapse_integration.py
index ae9fb93e..07e2308d 100644
--- a/tests/core/test_dash_collapse_integration.py
+++ b/tests/core/test_dash_collapse_integration.py
@@ -76,7 +76,7 @@ def server():
@pytest.fixture(autouse=True)
-def _clear_cancel():
+def clear_cancel():
was_set = DOWNLOAD_CANCELLED.is_set()
DOWNLOAD_CANCELLED.clear()
yield
@@ -86,16 +86,16 @@ def _clear_cancel():
DOWNLOAD_CANCELLED.clear()
-def _base(srv) -> str:
+def base(srv) -> str:
return f"http://127.0.0.1:{srv.server_address[1]}/"
-def _collapse_mpd(srv) -> str:
+def collapse_mpd(srv) -> str:
# SegmentList on one media.mp4: init is bytes [0, INIT_LEN), media ranges cover the rest
# contiguously -> collapsible_single_url returns the URL and download_track collapses.
return f"""
- {_base(srv)}
+ {base(srv)}
@@ -111,11 +111,11 @@ def _collapse_mpd(srv) -> str:
"""
-def _mixed_mpd(srv) -> str:
+def mixed_mpd(srv) -> str:
# Two SegmentURLs on distinct paths -> mixed URLs -> predicate False -> segmented path.
return f"""
- {_base(srv)}
+ {base(srv)}
@@ -130,11 +130,11 @@ def _mixed_mpd(srv) -> str:
"""
-def _track(mpd: str, srv):
- return DASH.from_text(mpd, url=f"{_base(srv)}manifest.mpd").to_tracks(language="en").videos[0]
+def make_track(mpd: str, srv):
+ return DASH.from_text(mpd, url=f"{base(srv)}manifest.mpd").to_tracks(language="en").videos[0]
-def _ctx(tmp_path, name="out.mp4"):
+def make_ctx(tmp_path, name="out.mp4"):
return DownloadContext(
save_path=tmp_path / name,
save_dir=tmp_path / "v0_segments",
@@ -144,8 +144,8 @@ def _ctx(tmp_path, name="out.mp4"):
def test_collapse_downloads_parent_whole_in_one_direct_request(server, tmp_path):
server.routes["/media.mp4"] = PARENT
- track = _track(_collapse_mpd(server), server)
- ctx = _ctx(tmp_path, "collapsed.mp4")
+ track = make_track(collapse_mpd(server), server)
+ ctx = make_ctx(tmp_path, "collapsed.mp4")
DASH.download_track(track, ctx)
@@ -166,8 +166,8 @@ def test_non_collapse_takes_segmented_merge_path(server, tmp_path):
server.routes["/init.bin"] = INIT2
server.routes["/part0.bin"] = PART0
server.routes["/part1.bin"] = PART1
- track = _track(_mixed_mpd(server), server)
- ctx = _ctx(tmp_path, "merged.mp4")
+ track = make_track(mixed_mpd(server), server)
+ ctx = make_ctx(tmp_path, "merged.mp4")
DASH.download_track(track, ctx)
diff --git a/tests/core/test_downloader_resume.py b/tests/core/test_downloader_resume.py
index 686ee15f..24333277 100644
--- a/tests/core/test_downloader_resume.py
+++ b/tests/core/test_downloader_resume.py
@@ -30,7 +30,7 @@ from requests import Session
dl = importlib.import_module("unshackle.core.downloaders.requests")
-def _consume(gen) -> list:
+def consume(gen) -> list:
return list(gen)
@@ -100,7 +100,7 @@ class _FaultServer:
@pytest.fixture(autouse=True)
-def _fast_retries(monkeypatch: pytest.MonkeyPatch) -> None:
+def fast_retries(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(dl, "RETRY_WAIT", 0)
@@ -112,7 +112,7 @@ def test_truncated_segment_is_not_finalized(tmp_path: Path, monkeypatch: pytest.
with _FaultServer(body) as srv:
srv.truncate_first = 99 # every attempt under-delivers
with pytest.raises(Exception):
- _consume(dl.download(url=srv.url, save_path=save, session=Session(), segmented=True))
+ consume(dl.download(url=srv.url, save_path=save, session=Session(), segmented=True))
assert not save.exists(), "truncated segment must not be finalized as complete"
@@ -123,7 +123,7 @@ def test_interrupted_download_resumes_from_bytes_written(tmp_path: Path) -> None
save = tmp_path / "file.bin"
with _FaultServer(body) as srv:
srv.truncate_first = 1 # first attempt delivers half then closes; retry completes
- _consume(dl.download(url=srv.url, save_path=save, session=Session()))
+ consume(dl.download(url=srv.url, save_path=save, session=Session()))
assert save.read_bytes() == body
# attempt 1 had no Range; attempt 2 must resume partway in, not from 0 or past the end
assert srv.ranges[0] is None
@@ -139,7 +139,7 @@ def test_resumes_from_preexisting_dev_partial(tmp_path: Path) -> None:
tmp = save.with_name("file.bin.!dev")
tmp.write_bytes(body[:1500]) # correct prefix the interrupted run flushed to disk
with _FaultServer(body) as srv:
- _consume(dl.download(url=srv.url, save_path=save, session=Session()))
+ consume(dl.download(url=srv.url, save_path=save, session=Session()))
assert save.read_bytes() == body
assert not tmp.exists()
assert srv.ranges[0] == "bytes=1500-", f"expected resume from 1500, got {srv.ranges[0]}"
@@ -198,7 +198,7 @@ def test_byte_range_slice_segments_never_probed_or_boosted(tmp_path: Path, monke
url = f"http://{host}:{port}/media.mp4"
urls = [{"url": url, "headers": {"Range": f"bytes={i * seg}-{(i + 1) * seg - 1}"}} for i in range(nseg)]
try:
- _consume(
+ consume(
dl.requests(
urls=urls,
output_dir=tmp_path,
@@ -226,6 +226,6 @@ def test_oversized_resume_416_restarts_clean(tmp_path: Path) -> None:
tmp = save.with_name(f"{save.name}.!dev")
tmp.write_bytes(b"X" * 4096) # poisoned partial, bigger than the server's body
with _FaultServer(body) as srv:
- _consume(dl.download(url=srv.url, save_path=save, session=Session()))
+ consume(dl.download(url=srv.url, save_path=save, session=Session()))
assert save.read_bytes() == body
assert srv.ranges[0] is not None and "4096" in srv.ranges[0] # first request over-ranged → 416
diff --git a/tests/downloaders/test_cancel_hardening.py b/tests/downloaders/test_cancel_hardening.py
index 17922fb8..95b3d215 100644
--- a/tests/downloaders/test_cancel_hardening.py
+++ b/tests/downloaders/test_cancel_hardening.py
@@ -81,13 +81,13 @@ def server():
@pytest.fixture(autouse=True)
-def _clear_cancel():
+def clear_cancel():
dl.DOWNLOAD_CANCELLED.clear()
yield
dl.DOWNLOAD_CANCELLED.clear()
-def _url(server: _Server) -> str:
+def url(server: _Server) -> str:
host, port = server.server_address
return f"http://{host}:{port}/seg.bin"
@@ -96,16 +96,16 @@ def test_dispatch_parts_does_not_finalize_on_global_cancel(server, tmp_path):
server.gate = threading.Event()
save_path = tmp_path / "big.bin"
- def _cancel_when_started() -> None:
+ def cancel_when_started() -> None:
assert server.started.wait(timeout=10)
dl.DOWNLOAD_CANCELLED.set()
server.gate.set()
- canceller = threading.Thread(target=_cancel_when_started)
+ canceller = threading.Thread(target=cancel_when_started)
canceller.start()
events = list(
dl.dispatch_parts(
- url=_url(server),
+ url=url(server),
save_path=save_path,
session=Session(),
total_size=len(BODY),
@@ -124,9 +124,9 @@ def test_backoff_wait_exits_promptly_on_batch_abort(server, tmp_path, monkeypatc
monkeypatch.setattr(dl, "retry_sleep", lambda exc, attempts: 30.0)
abort = threading.Event()
- def _consume() -> None:
+ def consume() -> None:
for _ in dl.download(
- url=_url(server),
+ url=url(server),
save_path=tmp_path / "seg.bin",
session=Session(),
segmented=True,
@@ -134,7 +134,7 @@ def test_backoff_wait_exits_promptly_on_batch_abort(server, tmp_path, monkeypatc
):
pass
- worker = threading.Thread(target=_consume)
+ worker = threading.Thread(target=consume)
start = time.monotonic()
worker.start()
time.sleep(0.5) # let the first attempt fail and enter the backoff wait
diff --git a/tests/downloaders/test_download_hardening.py b/tests/downloaders/test_download_hardening.py
index a15316f3..a79cd518 100644
--- a/tests/downloaders/test_download_hardening.py
+++ b/tests/downloaders/test_download_hardening.py
@@ -70,11 +70,11 @@ def server():
thread.join(timeout=5)
-def _url(srv, path="/parent.bin"):
+def url(srv, path="/parent.bin"):
return f"http://127.0.0.1:{srv.server_address[1]}{path}"
-def _run(srv, tmp_path, urls, **kwargs):
+def run(srv, tmp_path, urls, **kwargs):
for _ in dl.requests(urls, output_dir=tmp_path, filename="seg_{i:04}.bin", **kwargs):
pass
return sorted(tmp_path.glob("seg_*.bin"))
@@ -90,7 +90,7 @@ def test_adaptive_controller_starts_at_cap(server, tmp_path, monkeypatch):
return ctl
monkeypatch.setattr(dl, "AdaptiveWorkerController", recorder)
- files = _run(server, tmp_path, [{"url": _url(server)} for _ in range(8)], max_workers=4, adaptive=True)
+ files = run(server, tmp_path, [{"url": url(server)} for _ in range(8)], max_workers=4, adaptive=True)
assert len(files) == 8
assert seen == {"cap": 4, "start": 4}
@@ -100,7 +100,7 @@ def test_small_segments_never_probed(server, tmp_path, monkeypatch):
raise AssertionError("tail boost probed a small-segment batch")
monkeypatch.setattr(dl, "probe_ranged", no_probe)
- files = _run(server, tmp_path, [{"url": _url(server)} for _ in range(30)], max_workers=4, adaptive=True)
+ files = run(server, tmp_path, [{"url": url(server)} for _ in range(30)], max_workers=4, adaptive=True)
assert len(files) == 30
assert all(f.read_bytes() == PARENT for f in files)
@@ -108,9 +108,9 @@ def test_small_segments_never_probed(server, tmp_path, monkeypatch):
def test_range_item_ignored_by_server_fails_not_corrupts(server, tmp_path, monkeypatch):
server.honor_range = False
monkeypatch.setattr(dl, "RETRY_WAIT", 0.01)
- urls = [{"url": _url(server), "headers": {"Range": f"bytes={SLICE_START}-{SLICE_END}"}}]
+ urls = [{"url": url(server), "headers": {"Range": f"bytes={SLICE_START}-{SLICE_END}"}}]
with pytest.raises(Exception):
- _run(server, tmp_path, urls, max_workers=1)
+ run(server, tmp_path, urls, max_workers=1)
# nothing finalized: a 200 full-parent body must never become the segment
assert not list(tmp_path.glob("seg_*.bin"))
@@ -119,8 +119,8 @@ def test_range_item_spanning_whole_resource_accepts_200(server, tmp_path):
# RFC 9110 lets a server ignore Range and answer 200 with the whole resource; when the
# slice spans that resource the body is byte-identical to the 206 and must be kept
server.honor_range = False
- urls = [{"url": _url(server), "headers": {"Range": f"bytes=0-{len(PARENT) - 1}"}}]
- files = _run(server, tmp_path, urls, max_workers=1)
+ urls = [{"url": url(server), "headers": {"Range": f"bytes=0-{len(PARENT) - 1}"}}]
+ files = run(server, tmp_path, urls, max_workers=1)
assert files[0].read_bytes() == PARENT
assert len(server.requests) == 1
@@ -128,8 +128,8 @@ def test_range_item_spanning_whole_resource_accepts_200(server, tmp_path):
def test_range_item_retry_rewrites_slice(server, tmp_path, monkeypatch):
server.truncate_first = True
monkeypatch.setattr(dl, "RETRY_WAIT", 0.01)
- urls = [{"url": _url(server), "headers": {"Range": f"bytes={SLICE_START}-{SLICE_END}"}}]
- files = _run(server, tmp_path, urls, max_workers=1)
+ urls = [{"url": url(server), "headers": {"Range": f"bytes={SLICE_START}-{SLICE_END}"}}]
+ files = run(server, tmp_path, urls, max_workers=1)
assert files[0].read_bytes() == PARENT[SLICE_START : SLICE_END + 1]
# the retry must re-send the item's own Range, never a resume Range from the partial
ranges = [r.get("Range") for r in server.requests]
@@ -138,24 +138,24 @@ def test_range_item_retry_rewrites_slice(server, tmp_path, monkeypatch):
def test_segment_requests_ask_identity_encoding(server, tmp_path):
- _run(server, tmp_path, [{"url": _url(server)}], max_workers=1)
+ run(server, tmp_path, [{"url": url(server)}], max_workers=1)
assert all(r.get("Accept-Encoding") == "identity" for r in server.requests)
-def _exc_with_headers(headers):
+def exc_with_headers(headers):
exc = Exception("boom")
exc.response = SimpleNamespace(headers=headers)
return exc
def test_retry_sleep_honors_numeric_retry_after():
- assert dl.retry_sleep(_exc_with_headers({"Retry-After": "3"}), 1) == 3.0
- assert dl.retry_sleep(_exc_with_headers({"Retry-After": "9999"}), 1) == 60.0 # session.MAX_BACKOFF cap
+ assert dl.retry_sleep(exc_with_headers({"Retry-After": "3"}), 1) == 3.0
+ assert dl.retry_sleep(exc_with_headers({"Retry-After": "9999"}), 1) == 60.0 # session.MAX_BACKOFF cap
def test_retry_sleep_honors_http_date_retry_after():
when = format_datetime(datetime.now(timezone.utc) + timedelta(seconds=10), usegmt=True)
- assert 5 <= dl.retry_sleep(_exc_with_headers({"Retry-After": when}), 1) <= 10
+ assert 5 <= dl.retry_sleep(exc_with_headers({"Retry-After": when}), 1) <= 10
def test_retry_sleep_exponential_backoff_with_jitter():
@@ -166,7 +166,7 @@ def test_retry_sleep_exponential_backoff_with_jitter():
def test_retry_sleep_invalid_retry_after_falls_back():
- v = dl.retry_sleep(_exc_with_headers({"Retry-After": "soon"}), 1)
+ v = dl.retry_sleep(exc_with_headers({"Retry-After": "soon"}), 1)
assert dl.RETRY_WAIT <= v <= dl.RETRY_WAIT * 1.1
@@ -174,20 +174,20 @@ def test_retry_sleep_naive_http_date_treated_as_utc():
# "-0000" makes parsedate_to_datetime return a naive datetime; must clamp, not raise
when = format_datetime(datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(seconds=10))
assert when.endswith("-0000")
- v = dl.retry_sleep(_exc_with_headers({"Retry-After": when}), 1)
+ v = dl.retry_sleep(exc_with_headers({"Retry-After": when}), 1)
assert 5 <= v <= 10
def test_retry_sleep_nan_retry_after_falls_back():
# float("nan") parses; a non-finite wait must fall back to the exponential path
- v = dl.retry_sleep(_exc_with_headers({"Retry-After": "nan"}), 1)
+ v = dl.retry_sleep(exc_with_headers({"Retry-After": "nan"}), 1)
assert dl.RETRY_WAIT <= v <= dl.RETRY_WAIT * 1.1
def test_retry_sleep_honors_retry_after_on_wrapped_cause():
# RnetSession raises MaxRetriesError whose __cause__ HTTPError carries the response
exc = Exception("max retries exceeded")
- exc.__cause__ = _exc_with_headers({"Retry-After": "3"})
+ exc.__cause__ = exc_with_headers({"Retry-After": "3"})
assert dl.retry_sleep(exc, 1) == 3.0
@@ -196,7 +196,7 @@ def test_transient_429_with_retry_after_does_not_abort_track(server, tmp_path, m
dl.DOWNLOAD_CANCELLED.clear()
monkeypatch.setattr(dl, "RETRY_WAIT", 0.01)
server.fail_429_first = 2
- files = _run(server, tmp_path, [{"url": _url(server, f"/seg{i}.bin")} for i in range(3)], max_workers=1)
+ files = run(server, tmp_path, [{"url": url(server, f"/seg{i}.bin")} for i in range(3)], max_workers=1)
assert len(files) == 3
assert all(f.read_bytes() == PARENT for f in files)
assert not dl.DOWNLOAD_CANCELLED.is_set()
diff --git a/tests/downloaders/test_multiprocess.py b/tests/downloaders/test_multiprocess.py
index 54d67a4e..22b781ac 100644
--- a/tests/downloaders/test_multiprocess.py
+++ b/tests/downloaders/test_multiprocess.py
@@ -31,7 +31,7 @@ dl = importlib.import_module("unshackle.core.downloaders.requests")
SEG_REPEAT = 32 # 8-digit index * 32 = 256-byte body, distinct per index
-def _seg_body(index: int) -> bytes:
+def seg_body(index: int) -> bytes:
"""Segment body that encodes its own index, so misfiled striding is detectable."""
return (f"{index:08d}" * SEG_REPEAT).encode()
@@ -49,7 +49,7 @@ class _Handler(BaseHTTPRequestHandler):
self.send_header("Content-Length", "0")
self.end_headers()
return
- body = _seg_body(index)
+ body = seg_body(index)
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
@@ -78,7 +78,7 @@ def server():
@pytest.fixture(autouse=True)
-def _clear_cancel():
+def clear_cancel():
# spawned children carry their own flag, but the parent loop reads this process's flag;
# keep it clean so cancel state never leaks between tests
dl.DOWNLOAD_CANCELLED.clear()
@@ -86,14 +86,14 @@ def _clear_cancel():
dl.DOWNLOAD_CANCELLED.clear()
-def _url(srv, path):
+def url(srv, path):
return f"http://127.0.0.1:{srv.server_address[1]}{path}"
def test_multiprocess_stride_downloads_all_segments_correctly(server, tmp_path):
# >= MP_MIN_SEGMENTS with processes=2 engages the fan-out via the public entry point
n = dl.MP_MIN_SEGMENTS
- urls = [{"url": _url(server, f"/seg/{i}.bin")} for i in range(n)]
+ urls = [{"url": url(server, f"/seg/{i}.bin")} for i in range(n)]
total_advance = 0
for event in dl.requests(urls, output_dir=tmp_path, filename="seg_{i:04}.bin", max_workers=4, processes=2):
@@ -103,7 +103,7 @@ def test_multiprocess_stride_downloads_all_segments_correctly(server, tmp_path):
assert len(files) == n
for f in files:
index = int(f.stem.split("_")[1])
- assert f.read_bytes() == _seg_body(index), f"segment {index} holds another index's bytes (stride mapping bug)"
+ assert f.read_bytes() == seg_body(index), f"segment {index} holds another index's bytes (stride mapping bug)"
# parent drops child advances and emits one advance=1 per file_downloaded
assert total_advance == n
@@ -112,7 +112,7 @@ def test_multiprocess_child_crash_raises(server, tmp_path):
# download_multiprocess has no MP_MIN_SEGMENTS gate, so drive it directly with 2 urls.
# a dict url without "url" raises KeyError while requests() builds the save path (before
# any retry loop), so the child sends __mp_error__ fast; the sibling downloads normally.
- good = {"url": _url(server, "/seg/0.bin")}
+ good = {"url": url(server, "/seg/0.bin")}
bad = {"nourl": True}
with pytest.raises(RuntimeError):
for _ in dl.download_multiprocess(
@@ -134,7 +134,7 @@ def test_multiprocess_child_crash_raises(server, tmp_path):
def test_multiprocess_cancel_terminates_children(server, tmp_path):
server.delay = 0.5 # slow segments so cancel lands well before the batch could finish
n = dl.MP_MIN_SEGMENTS
- urls = [{"url": _url(server, f"/seg/{i}.bin")} for i in range(n)]
+ urls = [{"url": url(server, f"/seg/{i}.bin")} for i in range(n)]
baseline = len(multiprocessing.active_children())
start = time.time()
@@ -160,12 +160,12 @@ def test_speed_limit_forces_single_process(server, tmp_path, monkeypatch):
"""A set speed limit must keep the batch in-process: spawned children re-import the
module and would run with no TokenBucket, silently ignoring the configured cap."""
- def _no_mp(**kwargs):
+ def no_mp(**kwargs):
raise AssertionError("multiprocess fan-out engaged despite a speed limit")
- monkeypatch.setattr(dl, "download_multiprocess", _no_mp)
+ monkeypatch.setattr(dl, "download_multiprocess", no_mp)
n = dl.MP_MIN_SEGMENTS
- urls = [{"url": _url(server, f"/seg/{i}.bin")} for i in range(n)]
+ urls = [{"url": url(server, f"/seg/{i}.bin")} for i in range(n)]
dl.set_speed_limit(1_000_000_000) # far above the tiny batch, so the test is not slowed
try:
for _ in dl.requests(urls, output_dir=tmp_path, filename="seg_{i:04}.bin", max_workers=4, processes=2):
@@ -176,4 +176,4 @@ def test_speed_limit_forces_single_process(server, tmp_path, monkeypatch):
files = sorted(tmp_path.glob("seg_*.bin"))
assert len(files) == n
for f in files:
- assert f.read_bytes() == _seg_body(int(f.stem.split("_")[1]))
+ assert f.read_bytes() == seg_body(int(f.stem.split("_")[1]))
diff --git a/tests/downloaders/test_ranged_parallel.py b/tests/downloaders/test_ranged_parallel.py
index 4bfeb65b..b6372594 100644
--- a/tests/downloaders/test_ranged_parallel.py
+++ b/tests/downloaders/test_ranged_parallel.py
@@ -71,7 +71,7 @@ class _RangeHandler(BaseHTTPRequestHandler):
@pytest.fixture(autouse=True)
-def _clear_cancelled():
+def clear_cancelled():
# ranged failures must never touch the process-global flag; isolate every test from it
dl.DOWNLOAD_CANCELLED.clear()
yield
@@ -91,11 +91,11 @@ def server():
thread.join(timeout=5)
-def _url(srv, path="/payload.bin"):
+def url(srv, path="/payload.bin"):
return f"http://127.0.0.1:{srv.server_address[1]}{path}"
-def _run(srv, tmp_path, urls, **kwargs):
+def run(srv, tmp_path, urls, **kwargs):
for _ in dl.requests(urls, output_dir=tmp_path, filename="seg_{i:04}.bin", **kwargs):
pass
return sorted(tmp_path.glob("seg_*.bin"))
@@ -105,7 +105,7 @@ def test_ranged_parallel_merges_byte_identical(server, tmp_path, monkeypatch):
# shrink the thresholds so a 32 KiB payload splits into several parts
monkeypatch.setattr(dl, "RANGE_PARALLEL_MIN_SIZE", 8 * 1024)
monkeypatch.setattr(dl, "RANGE_PARALLEL_PART_SIZE", 2 * 1024)
- files = _run(server, tmp_path, [{"url": _url(server)}], max_workers=4)
+ files = run(server, tmp_path, [{"url": url(server)}], max_workers=4)
assert len(files) == 1
assert files[0].read_bytes() == PAYLOAD
# the split must actually have happened: >1 distinct window beyond the 0-0 probe
@@ -120,7 +120,7 @@ def test_ranged_part_requires_206(server, tmp_path, monkeypatch):
monkeypatch.setattr(dl, "RETRY_WAIT", 0.01)
save_path = tmp_path / "seg_0000.bin"
gen = dl.dispatch_parts(
- url=_url(server),
+ url=url(server),
save_path=save_path,
session=Session(),
total_size=len(PAYLOAD),
@@ -141,7 +141,7 @@ def test_ranged_part_failure_stays_local(server, tmp_path, monkeypatch):
dl.DOWNLOAD_CANCELLED.clear()
save_path = tmp_path / "seg_0000.bin"
gen = dl.dispatch_parts(
- url=_url(server),
+ url=url(server),
save_path=save_path,
session=Session(),
total_size=len(PAYLOAD),
@@ -157,8 +157,8 @@ def test_ranged_part_failure_stays_local(server, tmp_path, monkeypatch):
def test_probe_ranged_parses_content_range(server):
session = Session()
server.mode = "range"
- assert dl.probe_ranged(_url(server), session) == (len(PAYLOAD), True)
+ assert dl.probe_ranged(url(server), session) == (len(PAYLOAD), True)
server.mode = "always_200"
- assert dl.probe_ranged(_url(server), session) == (0, False)
+ assert dl.probe_ranged(url(server), session) == (0, False)
server.mode = "gzip_probe"
- assert dl.probe_ranged(_url(server), session) == (0, False)
+ assert dl.probe_ranged(url(server), session) == (0, False)
diff --git a/tests/downloaders/test_speed_limit.py b/tests/downloaders/test_speed_limit.py
index 18ce8b25..636ae691 100644
--- a/tests/downloaders/test_speed_limit.py
+++ b/tests/downloaders/test_speed_limit.py
@@ -13,7 +13,7 @@ requests_downloader = importlib.import_module("unshackle.core.downloaders.reques
@pytest.fixture(autouse=True)
-def _clean_cancel():
+def clean_cancel():
# a failing batch sets the process-global cancel, which TokenBucket.consume waits on;
# keep tests independent
DOWNLOAD_CANCELLED.clear()
diff --git a/tests/downloaders/test_tail_boost.py b/tests/downloaders/test_tail_boost.py
index c5c1f6c2..f7de6789 100644
--- a/tests/downloaders/test_tail_boost.py
+++ b/tests/downloaders/test_tail_boost.py
@@ -88,7 +88,7 @@ def server():
@pytest.fixture(autouse=True)
-def _clean_cancel():
+def clean_cancel():
# a failing batch sets the process-global cancel; keep tests independent
dl.DOWNLOAD_CANCELLED.clear()
yield
@@ -96,20 +96,20 @@ def _clean_cancel():
@pytest.fixture(autouse=True)
-def _boost_constants(monkeypatch):
+def boost_constants(monkeypatch):
monkeypatch.setattr(dl, "TAIL_BOOST_MIN_SEGMENT_SIZE", BOOST_MIN)
monkeypatch.setattr(dl, "TAIL_BOOST_PART_SIZE", BOOST_PART)
-def _url(srv, path):
+def url(srv, path):
return f"http://127.0.0.1:{srv.server_address[1]}{path}"
-def _urls(srv):
- return [{"url": _url(srv, f"/seg{i}.bin")} for i in range(SEG_COUNT)]
+def seg_urls(srv):
+ return [{"url": url(srv, f"/seg{i}.bin")} for i in range(SEG_COUNT)]
-def _run(srv, tmp_path, urls):
+def run(srv, tmp_path, urls):
advances = 0
for ev in dl.requests(urls, output_dir=tmp_path, filename="seg_{i:04}.bin", max_workers=MAX_WORKERS, adaptive=True):
a = ev.get("advance")
@@ -118,7 +118,7 @@ def _run(srv, tmp_path, urls):
return sorted(tmp_path.glob("seg_*.bin")), advances
-def _ranged_counts(srv):
+def ranged_counts(srv):
with srv.lock:
reqs = list(srv.requests)
counts: dict[str, int] = {}
@@ -128,15 +128,15 @@ def _ranged_counts(srv):
return counts
-def _boost_engaged(srv):
+def boost_engaged(srv):
# a boosted path sees a 0-0 probe plus >=2 part windows; a normal segment sees no Range
- return any(n > 1 for n in _ranged_counts(srv).values())
+ return any(n > 1 for n in ranged_counts(srv).values())
def test_tail_boost_split_produces_byte_identical_segment(server, tmp_path):
- files, advances = _run(server, tmp_path, _urls(server))
+ files, advances = run(server, tmp_path, seg_urls(server))
- assert _boost_engaged(server), "tail boost never engaged; test would silently cover the normal path"
+ assert boost_engaged(server), "tail boost never engaged; test would silently cover the normal path"
assert len(files) == SEG_COUNT
assert all(f.read_bytes() == BODY for f in files)
# part-mode byte advances are swallowed; each segment reports exactly one advance=1
@@ -147,13 +147,13 @@ def test_tail_boost_split_produces_byte_identical_segment(server, tmp_path):
def test_tail_boost_finalize_last_part_wins(server, tmp_path):
- files, _ = _run(server, tmp_path, _urls(server))
+ files, _ = run(server, tmp_path, seg_urls(server))
- assert _boost_engaged(server), "tail boost never engaged; finalize path untested"
+ assert boost_engaged(server), "tail boost never engaged; finalize path untested"
assert len(files) == SEG_COUNT
# last part wins: the pre-truncated .tp.!dev is os.replace'd into save_path, leaving no target
assert not list(tmp_path.glob("*.tp.!dev"))
- for path in _ranged_counts(server):
+ for path in ranged_counts(server):
i = int(path.removeprefix("/seg").removesuffix(".bin"))
save_path = tmp_path / f"seg_{i:04}.bin"
assert save_path.exists()
@@ -165,7 +165,7 @@ def test_tail_boost_part_failure_fails_batch_but_keeps_siblings_files_clean(serv
monkeypatch.setattr(dl, "RETRY_WAIT", 0.01)
with pytest.raises(Exception):
- _run(server, tmp_path, _urls(server))
+ run(server, tmp_path, seg_urls(server))
# the boosted tail segments (6, 7) hit the failing windows: no part run reaches parts_left==0
# cleanly, so neither is finalized. sibling leaders that did finalize must be uncorrupted.
diff --git a/tests/test_downloader_leaks.py b/tests/test_downloader_leaks.py
index d397e5a6..49580a69 100644
--- a/tests/test_downloader_leaks.py
+++ b/tests/test_downloader_leaks.py
@@ -87,16 +87,16 @@ def fast_timeouts() -> Iterator[None]:
DOWNLOAD_CANCELLED.clear()
-def _pool_threads() -> int:
+def pool_threads() -> int:
return sum(1 for t in threading.enumerate() if t.name.startswith("ThreadPoolExecutor"))
-def _assert_drained_and_unlocked(out_dir: Path, baseline: int) -> None:
+def assert_drained_and_unlocked(out_dir: Path, baseline: int) -> None:
"""Workers must exit within one read timeout; then every stray must be deletable."""
deadline = time.monotonic() + FAST["READ_TIMEOUT"] + FAST["RETRY_WAIT"] + 3
- while time.monotonic() < deadline and _pool_threads() > baseline:
+ while time.monotonic() < deadline and pool_threads() > baseline:
time.sleep(0.05)
- assert _pool_threads() <= baseline, "worker threads leaked past the drain deadline"
+ assert pool_threads() <= baseline, "worker threads leaked past the drain deadline"
for stray in out_dir.glob("*.!dev"):
stray.unlink() # PermissionError here on Windows = a handle is still held
# a resurrected worker would reopen/recreate its .!dev within one retry cycle
@@ -107,7 +107,7 @@ def _assert_drained_and_unlocked(out_dir: Path, baseline: int) -> None:
def test_failed_batch_frees_handles_even_after_flag_clear(server: _Server, tmp_path: Path) -> None:
"""Stall fault fails the batch; clearing DOWNLOAD_CANCELLED (as dl.py does for the
next track) must not resurrect the leaked racer's retry loop."""
- baseline = _pool_threads()
+ baseline = pool_threads()
host, port = str(server.server_address[0]), int(server.server_address[1])
urls = [f"http://{host}:{port}/stall/0"] + [f"http://{host}:{port}/ok/{i}" for i in range(1, 4)]
@@ -117,13 +117,13 @@ def test_failed_batch_frees_handles_even_after_flag_clear(server: _Server, tmp_p
assert DOWNLOAD_CANCELLED.is_set()
DOWNLOAD_CANCELLED.clear() # what dl.py does after a failed track
- _assert_drained_and_unlocked(tmp_path, baseline)
+ assert_drained_and_unlocked(tmp_path, baseline)
def test_external_cancel_frees_handles(server: _Server, tmp_path: Path) -> None:
"""DOWNLOAD_CANCELLED set mid-stream (sibling-track failure / SIGINT path): the batch
must wind down without leaving locked handles or live workers."""
- baseline = _pool_threads()
+ baseline = pool_threads()
host, port = str(server.server_address[0]), int(server.server_address[1])
urls = [f"http://{host}:{port}/slow/{i}" for i in range(4)]
@@ -132,4 +132,4 @@ def test_external_cancel_frees_handles(server: _Server, tmp_path: Path) -> None:
DOWNLOAD_CANCELLED.set()
DOWNLOAD_CANCELLED.clear()
- _assert_drained_and_unlocked(tmp_path, baseline)
+ assert_drained_and_unlocked(tmp_path, baseline)