162 lines
5.5 KiB
Python
162 lines
5.5 KiB
Python
"""Mises à jour : semver, détection Gitea (API tags), déclenchement Watchtower, endpoints."""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.config import get_settings
|
|
from app.services import update as update_service
|
|
from app.version import get_version
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_update_cache():
|
|
update_service.invalidate_cache()
|
|
yield
|
|
update_service.invalidate_cache()
|
|
|
|
|
|
def _patch_gitea(monkeypatch, url: str = "https://git.example", repo: str = "roman/ohm") -> None:
|
|
settings = get_settings()
|
|
monkeypatch.setattr(settings, "gitea_url", url)
|
|
monkeypatch.setattr(settings, "gitea_repo", repo)
|
|
|
|
|
|
# ---------------------------------------------------------------- semver / version
|
|
|
|
|
|
def test_parse_tag():
|
|
assert update_service.parse_tag("v0.2.1") == (0, 2, 1)
|
|
assert update_service.parse_tag("0.10.3") == (0, 10, 3)
|
|
assert update_service.parse_tag("v0.2") is None
|
|
assert update_service.parse_tag("v0.2.1-beta") is None
|
|
assert update_service.parse_tag("nimporte") is None
|
|
|
|
|
|
def test_is_newer():
|
|
assert update_service.is_newer("v0.2.0", "0.1.9")
|
|
assert update_service.is_newer("v1.0.0", "v0.99.99")
|
|
assert not update_service.is_newer("v0.1.0", "0.1.0")
|
|
assert not update_service.is_newer("v0.1.0", "dev") # courant non semver → jamais forcé
|
|
|
|
|
|
def test_get_version_env_override(monkeypatch):
|
|
monkeypatch.setenv("OHM_VERSION", "9.9.9")
|
|
assert get_version() == "9.9.9"
|
|
|
|
|
|
# ---------------------------------------------------------------- détection Gitea
|
|
|
|
|
|
async def test_fetch_latest_no_auth_header(monkeypatch):
|
|
"""Dépôt public : appel sans en-tête Authorization."""
|
|
_patch_gitea(monkeypatch)
|
|
|
|
class FakeResponse:
|
|
def raise_for_status(self) -> None:
|
|
pass
|
|
|
|
def json(self) -> list[dict]:
|
|
return [{"name": "v0.3.0"}]
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
async def fake_get(self: httpx.AsyncClient, url: str, headers: dict | None = None) -> FakeResponse:
|
|
calls.append((url, headers or {}))
|
|
return FakeResponse()
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "get", fake_get)
|
|
assert await update_service.fetch_latest_version(force=True) == "v0.3.0"
|
|
assert calls[0][1] == {} # aucun en-tête d'auth
|
|
|
|
|
|
async def test_fetch_latest_picks_highest_semver(monkeypatch):
|
|
"""Plus haut semver retenu + patchnote du tag, puis cache."""
|
|
_patch_gitea(monkeypatch)
|
|
|
|
class FakeResponse:
|
|
def raise_for_status(self) -> None:
|
|
pass
|
|
|
|
def json(self) -> list[dict]:
|
|
return [
|
|
{"name": "v0.1.0", "message": "ancien"},
|
|
{"name": "v0.2.3", "message": "correctifs\n"},
|
|
{"name": "v1.0.0-rc"},
|
|
{"name": "divers"},
|
|
]
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
async def fake_get(self: httpx.AsyncClient, url: str, headers: dict | None = None) -> FakeResponse:
|
|
calls.append((url, headers or {}))
|
|
return FakeResponse()
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "get", fake_get)
|
|
status = await update_service.status()
|
|
|
|
assert status["latest"] == "v0.2.3"
|
|
assert status["notes"] == "correctifs"
|
|
assert calls == [("https://git.example/api/v1/repos/roman/ohm/tags?limit=20", {})]
|
|
# puis servi par le cache (plus d'appel réseau)
|
|
assert await update_service.fetch_latest_version() == "v0.2.3"
|
|
assert len(calls) == 1
|
|
|
|
|
|
async def test_fetch_latest_network_error_degrades(monkeypatch):
|
|
_patch_gitea(monkeypatch)
|
|
|
|
async def fake_get(self: httpx.AsyncClient, url: str, headers: dict | None = None):
|
|
raise httpx.ConnectError("injoignable")
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "get", fake_get)
|
|
assert await update_service.fetch_latest_version(force=True) is None
|
|
|
|
|
|
# ---------------------------------------------------------------- déclenchement
|
|
|
|
|
|
async def test_trigger_update_requires_watchtower():
|
|
with pytest.raises(update_service.UpdateError, match="Watchtower"):
|
|
await update_service.trigger_update()
|
|
|
|
|
|
# ---------------------------------------------------------------- endpoints
|
|
|
|
|
|
async def test_version_endpoint_public(client):
|
|
r = await client.get("/api/version")
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["version"] == get_version() and data["name"]
|
|
|
|
|
|
async def test_update_requires_admin(client):
|
|
r = await client.get("/api/admin/update", follow_redirects=False)
|
|
assert r.status_code == 303 and r.headers["location"] == "/login"
|
|
|
|
|
|
async def test_update_status_defaults(client, admin_cookies, monkeypatch):
|
|
# Aucun appel réseau : la détection renvoie None (repo sans tag)
|
|
async def fake_fetch(*, force: bool = False):
|
|
return None
|
|
|
|
monkeypatch.setattr(update_service, "fetch_latest_version", fake_fetch)
|
|
|
|
r = await client.get("/api/admin/update", cookies=admin_cookies)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["latest"] is None and data["update_available"] is False
|
|
assert data["docker"] is False and data["notes"] is None
|
|
|
|
|
|
async def test_update_check_degrades_gracefully(client, admin_cookies, monkeypatch):
|
|
# Gitea injoignable : le check répond quand même (latest=None)
|
|
_patch_gitea(monkeypatch, url="https://git.inexistant", repo="roman/ohm")
|
|
r = await client.post("/api/admin/update/check", cookies=admin_cookies)
|
|
assert r.status_code == 200 and r.json()["latest"] is None
|
|
|
|
|
|
async def test_update_apply_without_docker_is_502(client, admin_cookies):
|
|
r = await client.post("/api/admin/update/apply", cookies=admin_cookies)
|
|
assert r.status_code == 502
|