40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
import pytest
|
|
import typer
|
|
|
|
from pyxray.cli.runtime import request_cancel, run_download_items
|
|
from pyxray.config import CoreConfig
|
|
from pyxray.libs.paths import WorkdirPaths
|
|
from pyxray.xray.downloader import DownloadManager
|
|
from pyxray.xray.manager import XrayManager
|
|
|
|
|
|
def make_manager(tmp_path) -> XrayManager:
|
|
return XrayManager(CoreConfig(), WorkdirPaths.from_value(tmp_path))
|
|
|
|
|
|
def test_request_cancel_writes_cancel_file(tmp_path) -> None:
|
|
manager = make_manager(tmp_path)
|
|
|
|
request_cancel(manager, emit=lambda message: None)
|
|
|
|
assert manager.paths.download_cancel_file.exists()
|
|
|
|
|
|
def test_run_download_items_rejects_existing_lock(tmp_path) -> None:
|
|
manager = make_manager(tmp_path)
|
|
manager.paths.ensure()
|
|
manager.paths.download_lock_file.write_text("locked", encoding="utf-8")
|
|
|
|
with pytest.raises(typer.Exit) as exc:
|
|
run_download_items(manager, [], force=False, proxy=None, emit=lambda *args, **kwargs: None)
|
|
|
|
assert exc.value.exit_code == 1
|
|
|
|
|
|
def test_xray_manager_reuses_download_manager_for_workdir(tmp_path) -> None:
|
|
first = make_manager(tmp_path)
|
|
second = make_manager(tmp_path)
|
|
|
|
assert isinstance(first.downloads, DownloadManager)
|
|
assert first.downloads is second.downloads
|