test: move-only download_from_url to stand-alone util file

Can be reviewed via --color-moved=dimmed-zebra
This commit is contained in:
MarcoFalke 2026-02-26 09:46:01 +01:00
parent fa0cc1c5a4
commit faf96286ce
No known key found for this signature in database
2 changed files with 50 additions and 40 deletions

47
test/download_utils.py Normal file
View File

@ -0,0 +1,47 @@
#!/usr/bin/env python3
#
# Copyright (c) The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or https://opensource.org/license/mit/.
import time
import urllib.request
def download_from_url(url, archive):
last_print_time = time.time()
def progress_hook(progress_bytes, total_size):
nonlocal last_print_time
now = time.time()
percent = min(100, (progress_bytes * 100) / total_size)
bar_length = 40
filled_length = int(bar_length * percent / 100)
bar = '#' * filled_length + '-' * (bar_length - filled_length)
if now - last_print_time >= 1 or percent >= 100:
print(f'\rDownloading: [{bar}] {percent:.1f}%', flush=True, end="")
last_print_time = now
with urllib.request.urlopen(url) as response:
if response.status != 200:
raise RuntimeError(f"HTTP request failed with status code: {response.status}")
sock_info = response.fp.raw._sock.getpeername()
print(f"Connected to {sock_info[0]}")
total_size = int(response.getheader("Content-Length"))
progress_bytes = 0
with open(archive, 'wb') as file:
while True:
chunk = response.read(8192)
if not chunk:
break
file.write(chunk)
progress_bytes += len(chunk)
progress_hook(progress_bytes, total_size)
if progress_bytes < total_size:
raise RuntimeError(f"Download incomplete: expected {total_size} bytes, got {progress_bytes} bytes")
print('\n', flush=True, end="") # Flush to avoid error output on the same line.

View File

@ -16,9 +16,11 @@ import shutil
import subprocess
import sys
import time
import urllib.request
import zipfile
sys.path.append(str(Path(__file__).resolve().parent))
from download_utils import download_from_url
TAR = os.getenv('TAR', 'tar')
SHA256_SUMS = {
@ -102,45 +104,6 @@ def pushd(new_dir) -> None:
os.chdir(previous_dir)
def download_from_url(url, archive):
last_print_time = time.time()
def progress_hook(progress_bytes, total_size):
nonlocal last_print_time
now = time.time()
percent = min(100, (progress_bytes * 100) / total_size)
bar_length = 40
filled_length = int(bar_length * percent / 100)
bar = '#' * filled_length + '-' * (bar_length - filled_length)
if now - last_print_time >= 1 or percent >= 100:
print(f'\rDownloading: [{bar}] {percent:.1f}%', flush=True, end="")
last_print_time = now
with urllib.request.urlopen(url) as response:
if response.status != 200:
raise RuntimeError(f"HTTP request failed with status code: {response.status}")
sock_info = response.fp.raw._sock.getpeername()
print(f"Connected to {sock_info[0]}")
total_size = int(response.getheader("Content-Length"))
progress_bytes = 0
with open(archive, 'wb') as file:
while True:
chunk = response.read(8192)
if not chunk:
break
file.write(chunk)
progress_bytes += len(chunk)
progress_hook(progress_bytes, total_size)
if progress_bytes < total_size:
raise RuntimeError(f"Download incomplete: expected {total_size} bytes, got {progress_bytes} bytes")
print('\n', flush=True, end="") # Flush to avoid error output on the same line.
def download_binary(tag, args) -> int:
if Path(tag).is_dir():
if not args.remove_dir: