ohm_streaming/tests/test_update.py

"""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