fix(updater): infinite loop when fetch url does not depend on version

Author: fyhertz

src/ops2deb/updater.py

@@ -19,9 +19,16 @@ class BaseUpdateStrategy:
     def __init__(self, client: httpx.AsyncClient):
         self.client = client
 
-    async def _try_version(self, blueprint: Blueprint, version: str) -> bool:
-        if not (url := blueprint.render_fetch_url(version=version)):
+    async def try_version(self, blueprint: Blueprint, version: str) -> bool:
+        # No need to waste an HTTP call when called with the current blueprint version
+        if blueprint.version == version:
+            return True
+
+        # Fetch url does not depend on blueprint version or blueprint has no fetch
+        url = blueprint.render_fetch_url(version=version)
+        if url == blueprint.render_fetch_url() or url is None:
             return False
+
         logger.debug(f"{self.__class__.__name__} - {blueprint.name} - Trying {url}")
         try:
             response = await self.client.head(url)
@@ -53,7 +60,7 @@ async def _try_a_few_patches(
     ) -> Version | None:
         for i in range(0, 3):
             version = version.bump_patch()
-            if await self._try_version(blueprint, str(version)) is True:
+            if await self.try_version(blueprint, str(version)) is True:
                 return version
         return None
 
@@ -64,7 +71,7 @@ async def _try_versions(
         version_part: str,
     ) -> Version:
         bumped_version = getattr(version, f"bump_{version_part}")()
-        if await self._try_version(blueprint, str(bumped_version)) is False:
+        if await self.try_version(blueprint, str(bumped_version)) is False:
             if version_part != "patch":
                 if (
                     result := await self._try_a_few_patches(blueprint, bumped_version)
@@ -145,7 +152,7 @@ async def __call__(self, blueprint: Blueprint) -> str:
         version = tag_name if not tag_name.startswith("v") else tag_name[1:]
         if Version.isvalid(version) and Version.isvalid(blueprint.version):
             version = str(max(Version.parse(version), Version.parse(blueprint.version)))
-        if await self._try_version(blueprint, version) is False:
+        if await self.try_version(blueprint, version) is False:
             raise Ops2debUpdaterError(
                 f"Failed to determine latest release URL (latest tag is {tag_name})"
             )

tests/test_updater.py

@@ -1,13 +1,18 @@
 from typing import Optional
 
+import httpx
 import pytest
 from fastapi import FastAPI, HTTPException
 from httpx import AsyncClient
 from starlette.responses import JSONResponse
 
 from ops2deb.exceptions import Ops2debUpdaterError
 from ops2deb.logger import enable_debug
-from ops2deb.updater import GenericUpdateStrategy, GithubUpdateStrategy
+from ops2deb.updater import (
+    BaseUpdateStrategy,
+    GenericUpdateStrategy,
+    GithubUpdateStrategy,
+)
 
 enable_debug(True)
 
@@ -47,6 +52,23 @@ def github_release_api():
     return _github_app_factory
 
 
+async def test_try_version__returns_false_when_fetch_url_does_not_depend_on_the_blueprint_version(  # noqa: E501
+    blueprint_factory,
+):
+    # Given
+    blueprint = blueprint_factory(
+        version="1.0.0",
+        fetch="http://test/releases/1.0.0/some-app.tar.gz",
+    )
+    update_strategy = BaseUpdateStrategy(httpx.AsyncClient())
+
+    # When
+    result = await update_strategy.try_version(blueprint, "2.0.0")
+
+    # Then
+    assert result is False
+
+
 @pytest.mark.parametrize(
     "versions,expected_result",
     [
@@ -60,16 +82,42 @@ def github_release_api():
         (["1.0.0", "1.0.1", "1.0.2", "1.1.0", "1.1.1"], "1.1.1"),
     ],
 )
-async def test_generic_update_strategy_should_find_expected_blueprint_release(
+async def test_generic_update_strategy_finds_latest_release_version(
     blueprint_factory, app_factory, versions, expected_result
 ):
+    # Given
     blueprint = blueprint_factory(
+        version="1.0.0",
         fetch="http://test/releases/{{version}}/some-app.tar.gz",
     )
     app = app_factory(versions)
+
+    # When
+    async with AsyncClient(app=app) as client:
+        update_strategy = GenericUpdateStrategy(client)
+        latest_version = await update_strategy(blueprint)
+
+    # Then
+    assert latest_version == expected_result
+
+
+async def test_generic_update_strategy_finds_latest_release_version_when_version_has_prerelease_part(  # noqa: E501
+    blueprint_factory, app_factory
+):
+    # Given
+    blueprint = blueprint_factory(
+        version="1.0.0-pre",
+        fetch="http://test/releases/{{version}}/some-app.tar.gz",
+    )
+    app = app_factory(["2.0.0"])
+
+    # When
     async with AsyncClient(app=app) as client:
         update_strategy = GenericUpdateStrategy(client)
-        assert await update_strategy(blueprint) == expected_result
+        latest_version = await update_strategy(blueprint)
+
+    # Then
+    assert latest_version == "2.0.0"
 
 
 @pytest.mark.parametrize(