80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from pyxray.api.app import create_app
|
|
|
|
|
|
def test_assets_config_returns_core_toml_text(tmp_path) -> None:
|
|
client = TestClient(create_app(tmp_path))
|
|
|
|
response = client.get("/api/v1/assets/config")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["path"].endswith("core.toml")
|
|
assert "[download]" in payload["content"]
|
|
assert "workdir" not in payload["content"]
|
|
|
|
|
|
def test_assets_status_reports_missing_assets(tmp_path) -> None:
|
|
client = TestClient(create_app(tmp_path))
|
|
|
|
response = client.get("/api/v1/assets/status")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["installed"] is False
|
|
assert payload["geoip"]["exists"] is False
|
|
assert payload["geosite"]["exists"] is False
|
|
|
|
|
|
def test_assets_download_without_urls_starts_default_xray_task(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
"pyxray.xray.manager.default_xray_url",
|
|
lambda: "https://default.invalid/xray.zip",
|
|
)
|
|
client = TestClient(create_app(tmp_path))
|
|
|
|
response = client.post("/api/v1/assets/download", json={"force": True})
|
|
|
|
assert response.status_code == 202
|
|
payload = response.json()
|
|
assert payload["task_id"] == "current"
|
|
assert payload["state"] == "running"
|
|
assert payload["items"] == ["xray"]
|
|
|
|
|
|
def test_assets_download_rejects_when_lock_exists(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
"pyxray.xray.manager.default_xray_url",
|
|
lambda: "https://default.invalid/xray.zip",
|
|
)
|
|
app = create_app(tmp_path)
|
|
service = app.state.assets_service
|
|
service.manager.paths.ensure()
|
|
service.manager.paths.download_lock_file.write_text("locked", encoding="utf-8")
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/api/v1/assets/download", json={"force": True})
|
|
|
|
assert response.status_code == 409
|
|
assert response.json() == {"error": "another download is already running"}
|
|
|
|
|
|
def test_assets_download_task_returns_status(tmp_path) -> None:
|
|
client = TestClient(create_app(tmp_path))
|
|
|
|
response = client.get("/api/v1/assets/download/task")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] == "idle"
|
|
|
|
|
|
def test_assets_cancel_writes_cancel_signal(tmp_path) -> None:
|
|
app = create_app(tmp_path)
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/api/v1/assets/download/cancel")
|
|
|
|
assert response.status_code == 200
|
|
assert app.state.assets_service.manager.paths.download_cancel_file.exists()
|