Use device-reported repo for firmware releases and upgrades

src/wled/models.py

@@ -14,6 +14,7 @@
 from mashumaro.types import SerializableType, SerializationStrategy
 
 from .const import (
+    DEFAULT_REPO,
     MIN_REQUIRED_VERSION,
     LightCapability,
     LiveDataOverride,
@@ -489,6 +490,11 @@ class Info(BaseModel):  # pylint: disable=too-many-instance-attributes
     product: str = "DIY Light"
     """The product name. Always FOSS for standard installations."""
 
+    repo: str = field(default=DEFAULT_REPO, metadata=field_options(alias="repo"))
+    """GitHub repository in 'org/repo' format reported by the device firmware.
+    Used in preference to the default repo during upgrade operations.
+    """
+
     release: str | None = None
     """The release name of the firmware build.
 

src/wled/wled.py

@@ -654,6 +654,12 @@ async def upgrade(  # noqa: PLR0912
             msg = "Device already running the requested version"
             raise WLEDUpgradeError(msg)
 
+        # Prefer the repo reported by the device itself only when the caller
+        # did not override the default repository. Older firmware that does not
+        # include the field will already have fallen back to DEFAULT_REPO.
+        if repo == DEFAULT_REPO:
+            repo = self._device.info.repo
+
         # Determine if this is an Ethernet board
         ethernet = ""
         if (

tests/const.py

@@ -0,0 +1,5 @@
+"""Constants for tests."""
+
+from wled.const import DEFAULT_REPO
+
+__all__ = ["DEFAULT_REPO"]

tests/test_models.py

@@ -23,6 +23,7 @@
 from wled.utils import get_awesome_version
 
 from .conftest import full_device_data, load_fixture_json
+from .const import DEFAULT_REPO
 
 # =========================================================================
 # Helper functions
@@ -315,6 +316,18 @@ def test_info_version_deserialized() -> None:
     assert str(info.version) == "0.14.0"
 
 
+def test_info_repo_defaults_to_default_repo() -> None:
+    """Test repo defaults to DEFAULT_REPO when not present in response."""
+    info = Info.from_dict(_base_info())
+    assert info.repo == DEFAULT_REPO
+
+
+def test_info_repo_uses_device_value_when_present() -> None:
+    """Test repo is taken from the device response when present."""
+    info = Info.from_dict(_base_info(repo="MoonModules/WLED"))
+    assert info.repo == "MoonModules/WLED"
+
+
 # =========================================================================
 # State model
 # =========================================================================

tests/test_wled.py

@@ -1332,6 +1332,42 @@ async def test_upgrade_success() -> None:
             await wled.upgrade(version="0.15.0")
 
 
+async def test_upgrade_uses_device_repo() -> None:
+    """Test upgrade downloads firmware from the repo reported by the device."""
+    with aioresponses() as mocked:
+        async with aiohttp.ClientSession() as session:
+            wled_data = load_fixture_json("wled")
+            wled_data["info"]["arch"] = "esp32"
+            wled_data["info"]["ver"] = "0.14.0"
+            wled_data["info"]["repo"] = "MoonModules/WLED"
+            mocked.get(
+                "http://example.com/json",
+                status=200,
+                body=json.dumps(wled_data),
+                content_type="application/json",
+            )
+            mocked.get(
+                "http://example.com/presets.json",
+                status=200,
+                body=json.dumps(load_fixture_json("presets")),
+                content_type="application/json",
+            )
+            wled = WLED("example.com", session=session)
+            await wled.update()
+            mocked.get(
+                "https://github.com/MoonModules/WLED/releases/download/v0.15.0/WLED_0.15.0_ESP32.bin",
+                status=200,
+                body=b"fake firmware",
+            )
+            mocked.post(
+                "http://example.com/update",
+                status=200,
+                body="OK",
+                content_type="text/plain",
+            )
+            await wled.upgrade(version="0.15.0")
+
+
 async def test_upgrade_ethernet_board() -> None:
     """Test upgrade with Ethernet board (empty bssid)."""
     with aioresponses() as mocked: