From e3c1ce351192e4d594d851d7c084de2b409db743 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 16 Aug 2026 12:52:36 -0600 Subject: [PATCH] refactor(tests): drop leading underscores from test helper functions Completes the core underscore sweep for the test files this branch added; make_ctx/make_track/seg_urls avoid shadowing locals. --- tests/core/test_adaptive_workers.py | 32 ++++++++-------- tests/core/test_dash_collapse_integration.py | 26 ++++++------- tests/core/test_downloader_resume.py | 14 +++---- tests/downloaders/test_cancel_hardening.py | 16 ++++---- tests/downloaders/test_download_hardening.py | 40 ++++++++++---------- tests/downloaders/test_multiprocess.py | 24 ++++++------ tests/downloaders/test_ranged_parallel.py | 18 ++++----- tests/downloaders/test_speed_limit.py | 2 +- tests/downloaders/test_tail_boost.py | 30 +++++++-------- tests/test_downloader_leaks.py | 16 ++++---- 10 files changed, 109 insertions(+), 109 deletions(-) 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)