Roman
9148b5fb6a
- Détection de version via l'API Gitea, application via Watchtower - Router système, UI admin (intégrations), bandeau de mise à jour - Dockerfile multi-étapes, docker-compose, scripts/release.sh - Version centralisée dans app/version.py (pyproject ou OHM_VERSION au build)
185 lines
6.4 KiB
Python
185 lines
6.4 KiB
Python
"""Mises à jour : semver, détection Gitea (API tags), déclenchement Watchtower, endpoints."""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.services import update as update_service
|
|
from app.services.settings import get_update_config
|
|
from app.version import get_version
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_update_cache():
|
|
update_service.invalidate_cache()
|
|
yield
|
|
update_service.invalidate_cache()
|
|
|
|
|
|
# ---------------------------------------------------------------- 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_without_repo():
|
|
from app.services.settings import set_update_config
|
|
|
|
await set_update_config("https://git.example", "", "tok")
|
|
assert await update_service.fetch_latest_version(force=True) is None
|
|
|
|
|
|
async def test_fetch_latest_public_repo_no_token(monkeypatch):
|
|
"""Dépôt public : pas de jeton → appel sans en-tête Authorization."""
|
|
from app.services.settings import set_update_config
|
|
|
|
await set_update_config("https://git.example", "roman/ohm", "")
|
|
|
|
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):
|
|
"""Avec jeton : en-tête Authorization + plus haut semver retenu, puis cache."""
|
|
from app.services.settings import set_update_config
|
|
|
|
await set_update_config("https://git.example", "roman/ohm", "tok")
|
|
|
|
class FakeResponse:
|
|
def raise_for_status(self) -> None:
|
|
pass
|
|
|
|
def json(self) -> list[dict]:
|
|
return [{"name": "v0.1.0"}, {"name": "v0.2.3"}, {"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)
|
|
latest = await update_service.fetch_latest_version(force=True)
|
|
|
|
assert latest == "v0.2.3"
|
|
assert calls == [
|
|
("https://git.example/api/v1/repos/roman/ohm/tags?limit=20", {"Authorization": "token tok"})
|
|
]
|
|
# 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):
|
|
from app.services.settings import set_update_config
|
|
|
|
await set_update_config("https://git.example", "roman/ohm", "tok")
|
|
|
|
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["configured"] is True and data["docker"] is False # repo par défaut présent
|
|
assert data["latest"] is None and data["update_available"] is False
|
|
assert data["config"]["gitea_url"] == "https://git.lanro.eu"
|
|
assert data["config"]["repo"] == "Roman/ohm_streaming"
|
|
|
|
|
|
async def test_update_save_config_normalizes(client, admin_cookies):
|
|
r = await client.put(
|
|
"/api/admin/update",
|
|
json={"gitea_url": "https://git.example/", "repo": "/roman/ohm/", "token": " tok "},
|
|
cookies=admin_cookies,
|
|
)
|
|
assert r.status_code == 200
|
|
assert await get_update_config() == {
|
|
"gitea_url": "https://git.example",
|
|
"repo": "roman/ohm",
|
|
"token": "tok",
|
|
}
|
|
|
|
|
|
async def test_update_check_degrades_gracefully(client, admin_cookies):
|
|
# Gitea configuré mais injoignable : le check répond quand même (latest=None)
|
|
await client.put(
|
|
"/api/admin/update",
|
|
json={"gitea_url": "https://git.inexistant", "repo": "roman/ohm", "token": "tok"},
|
|
cookies=admin_cookies,
|
|
)
|
|
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
|