135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
from pathlib import Path
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from pyxray.config import CoreConfig
|
|
from pyxray.libs.paths import WorkdirPaths
|
|
from pyxray.xray.models import FileStatus, RemoteVersion, XrayStatus
|
|
from pyxray.xray.workflows import build_download_items, select_update_items, version_tuple
|
|
|
|
|
|
class FakeManager(BaseModel):
|
|
config: CoreConfig
|
|
paths: WorkdirPaths
|
|
local_status: XrayStatus
|
|
remote: RemoteVersion | None = None
|
|
|
|
model_config = {"arbitrary_types_allowed": True}
|
|
|
|
def status(self) -> XrayStatus:
|
|
return self.local_status
|
|
|
|
def effective_xray_url(self) -> str:
|
|
return self.config.xray.xray_url or "https://default.invalid/xray.zip"
|
|
|
|
def remote_version(self, proxy_url: str | None = None) -> RemoteVersion | None:
|
|
return self.remote
|
|
|
|
def download_xray(self, url: str, proxy_url: str | None, cancel_file: Path):
|
|
raise NotImplementedError
|
|
|
|
def download_geo_bundle(self, url: str, proxy_url: str | None, cancel_file: Path):
|
|
raise NotImplementedError
|
|
|
|
def download_geoip(self, url: str, proxy_url: str | None, cancel_file: Path):
|
|
raise NotImplementedError
|
|
|
|
def download_geosite(self, url: str, proxy_url: str | None, cancel_file: Path):
|
|
raise NotImplementedError
|
|
|
|
|
|
def make_manager(
|
|
tmp_path,
|
|
*,
|
|
installed: bool = False,
|
|
version: str = "",
|
|
geoip: bool = False,
|
|
geosite: bool = False,
|
|
remote_version: str = "",
|
|
) -> FakeManager:
|
|
config = CoreConfig.model_validate(
|
|
{
|
|
"xray": {
|
|
"xray_url": "https://example.invalid/xray.zip",
|
|
"geo_url": "https://example.invalid/geo.zip",
|
|
"geoip_url": "https://example.invalid/geoip.dat",
|
|
"geosite_url": "https://example.invalid/geosite.dat",
|
|
}
|
|
}
|
|
)
|
|
status = XrayStatus(
|
|
installed=installed,
|
|
executable_path=str(tmp_path / "xray.exe"),
|
|
version=version,
|
|
healthy=bool(version),
|
|
geoip=FileStatus(exists=geoip, path=str(tmp_path / "geoip.dat")),
|
|
geosite=FileStatus(exists=geosite, path=str(tmp_path / "geosite.dat")),
|
|
)
|
|
remote = (
|
|
RemoteVersion(source="test", version=remote_version)
|
|
if remote_version
|
|
else None
|
|
)
|
|
return FakeManager(
|
|
config=config,
|
|
paths=WorkdirPaths.from_value(tmp_path),
|
|
local_status=status,
|
|
remote=remote,
|
|
)
|
|
|
|
|
|
def test_build_download_items_uses_explicit_urls(tmp_path) -> None:
|
|
manager = make_manager(tmp_path)
|
|
|
|
items = build_download_items(manager, xray_url="https://override.invalid/xray.zip")
|
|
|
|
assert items[0].name == "xray"
|
|
assert items[0].url == "https://override.invalid/xray.zip"
|
|
assert items[1].name == "geo"
|
|
|
|
|
|
def test_build_download_items_uses_default_xray_url_when_config_is_empty(
|
|
tmp_path,
|
|
) -> None:
|
|
manager = make_manager(tmp_path)
|
|
manager.config.xray.xray_url = ""
|
|
|
|
items = build_download_items(manager)
|
|
|
|
assert items[0].url == "https://default.invalid/xray.zip"
|
|
|
|
|
|
def test_update_selects_missing_xray_and_geo_files(tmp_path) -> None:
|
|
manager = make_manager(tmp_path, installed=False, geoip=False, geosite=False)
|
|
|
|
selected = select_update_items(manager, force=False, proxy_url=None)
|
|
|
|
assert [item.name for item in selected] == ["xray", "geo", "geoip", "geosite"]
|
|
|
|
|
|
def test_update_selects_newer_remote_xray(tmp_path) -> None:
|
|
manager = make_manager(
|
|
tmp_path,
|
|
installed=True,
|
|
version="1.0.0",
|
|
geoip=True,
|
|
geosite=True,
|
|
remote_version="1.1.0",
|
|
)
|
|
|
|
selected = select_update_items(manager, force=False, proxy_url=None)
|
|
|
|
assert [item.name for item in selected] == ["xray"]
|
|
|
|
|
|
def test_force_update_selects_every_configured_item(tmp_path) -> None:
|
|
manager = make_manager(tmp_path, installed=True, version="1.0.0", geoip=True, geosite=True)
|
|
|
|
selected = select_update_items(manager, force=True, proxy_url=None)
|
|
|
|
assert [item.name for item in selected] == ["xray", "geo", "geoip", "geosite"]
|
|
|
|
|
|
def test_version_tuple_extracts_semver() -> None:
|
|
assert version_tuple("v25.3.6") == (25, 3, 6)
|