mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-11-14 04:48:27 +00:00
misko903andClaude Opus 5.5
d50440f6c2
## What
A Webshare.cz download shows "Resolving Webshare link" for its entire
duration, so a large download (e.g. a 1.3 GB audiobook archive) looks
stuck.
## Why
After the link was resolved, the handler called
`status_callback("downloading", None)`. `update_download_status` treats
a `None` message as "keep the current message", so the "Resolving
Webshare link" detail stayed in the activity panel. The HTTP handler
sends `""` for exactly this reason.
## How
- Report `downloading` with an empty message, which clears the resolving
text.
- When the download response has no `Content-Length`, use the release's
size (`task.size`, parsed with `parse_size_string`) as the total for
progress. Without it, progress never moved, and stall detection would
flag a healthy long download after 5 minutes.
## Testing
- New tests in `tests/webshare/test_handler.py`: the downloading status
clears the message, and progress is computed from the release size when
there is no Content-Length.
- `uv run pytest tests/webshare` is green (34 passed), and `ruff check`
and `ruff format --check` are clean.
Follow-up to #1543.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""Tests for WebshareHandler: link resolution, streaming, cleanup on failure."""
|
|
|
|
import threading
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from shelfmark.core.models import DownloadTask
|
|
from shelfmark.release_sources.webshare import handler as webshare_handler
|
|
from shelfmark.release_sources.webshare.api import WebshareError
|
|
from shelfmark.release_sources.webshare.handler import WebshareHandler
|
|
|
|
|
|
def _task(task_id="webshare:abc123", fmt="epub"):
|
|
return DownloadTask(task_id=task_id, source="webshare", title="Krysar", format=fmt)
|
|
|
|
|
|
def _http_response(body: bytes, *, content_length=True):
|
|
response = MagicMock()
|
|
response.__enter__.return_value = response
|
|
response.__exit__.return_value = False
|
|
response.headers = {"Content-Length": str(len(body))} if content_length else {}
|
|
response.raise_for_status.return_value = None
|
|
response.iter_content.return_value = [body[i : i + 4096] for i in range(0, len(body), 4096)]
|
|
return response
|
|
|
|
|
|
def _run(tmp_path, *, file_link, body=b"x" * 20000, cancel=None, task=None, content_length=True):
|
|
status = MagicMock()
|
|
progress = MagicMock()
|
|
cancel = cancel or threading.Event()
|
|
with (
|
|
patch.object(webshare_handler, "TMP_DIR", tmp_path),
|
|
patch.object(
|
|
webshare_handler.config,
|
|
"get",
|
|
side_effect=lambda k, d=None: "none" if k == "FILE_ORGANIZATION" else d,
|
|
),
|
|
patch.object(webshare_handler, "get_credentials", return_value=("user", "pw")),
|
|
patch.object(webshare_handler.WebshareClient, "file_link", side_effect=file_link),
|
|
patch.object(
|
|
webshare_handler.requests,
|
|
"get",
|
|
return_value=_http_response(body, content_length=content_length),
|
|
),
|
|
):
|
|
result = WebshareHandler().download(task or _task(), cancel, progress, status)
|
|
return result, status, progress
|
|
|
|
|
|
def test_download_streams_file_named_after_link(tmp_path):
|
|
idents = []
|
|
|
|
def file_link(ident):
|
|
idents.append(ident)
|
|
return "https://dl.webshare.cz/abc/Krysar%20%28CZ%29.epub"
|
|
|
|
result, _, progress = _run(tmp_path, file_link=file_link)
|
|
|
|
expected = tmp_path / "Krysar (CZ).epub"
|
|
assert result == str(expected)
|
|
assert expected.read_bytes() == b"x" * 20000
|
|
assert idents == ["abc123"]
|
|
progress.assert_called_with(100.0)
|
|
|
|
|
|
def test_downloading_status_clears_resolving_message(tmp_path):
|
|
_, status, _ = _run(tmp_path, file_link=lambda ident: "https://dl.webshare.cz/f.epub")
|
|
|
|
# None would keep "Resolving Webshare link" on screen for the whole download.
|
|
status.assert_any_call("downloading", "")
|
|
|
|
|
|
def test_progress_uses_release_size_without_content_length(tmp_path):
|
|
task = _task()
|
|
task.size = "40 KB"
|
|
_, _, progress = _run(
|
|
tmp_path,
|
|
file_link=lambda ident: "https://dl.webshare.cz/f.epub",
|
|
body=b"x" * 20480,
|
|
task=task,
|
|
content_length=False,
|
|
)
|
|
|
|
progress.assert_any_call(50.0)
|
|
|
|
|
|
def test_too_small_file_is_rejected_and_removed(tmp_path):
|
|
result, status, _ = _run(
|
|
tmp_path, file_link=lambda ident: "https://dl.webshare.cz/f.epub", body=b"tiny"
|
|
)
|
|
assert result is None
|
|
assert not (tmp_path / "f.epub").exists()
|
|
status.assert_called_with("error", "Webshare returned an invalid file")
|
|
|
|
|
|
def test_link_error_reports_message(tmp_path):
|
|
def file_link(ident):
|
|
msg = "File not found."
|
|
raise WebshareError(msg, "FILE_LINK_FATAL_1")
|
|
|
|
result, status, _ = _run(tmp_path, file_link=file_link)
|
|
assert result is None
|
|
status.assert_called_with("error", "File not found.")
|
|
|
|
|
|
def test_cancel_before_start(tmp_path):
|
|
cancel = threading.Event()
|
|
cancel.set()
|
|
file_link = MagicMock()
|
|
result, status, _ = _run(tmp_path, file_link=file_link, cancel=cancel)
|
|
assert result is None
|
|
file_link.assert_not_called()
|
|
status.assert_called_with("cancelled", "Cancelled")
|