From 1cd7fbd828222f8d440f552e595a29b68641b404 Mon Sep 17 00:00:00 2001 From: blueleo076 Date: Mon, 15 Jun 2026 07:27:21 +0200 Subject: [PATCH 1/3] feat(login): add candidate URL lists for new/legacy PMS firmware Newer LG ESS Home firmware (PMS 10.05.7438 and above) exposes the login endpoint at /v1/login and the timesync endpoint at /v1/timesync instead of the original /v1/user/setting/login and /v1/user/setting/timesync. Add LOGIN_URLS and TIMESYNC_URLS as ordered lists so aio_ess can try the new endpoint first and fall back to the legacy one when the device still speaks the old API. Keep the module-level LOGIN_URL and TIMESYNC_URL constants pointing at the legacy URLs for backward compatibility with callers that import them directly. Refs: https://github.com/gluap/pyess/issues/37 --- pyess/constants.py | 55 ++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 51 insertions(+), 4 deletions(-) diff --git a/pyess/constants.py b/pyess/constants.py index 87f138f..dbe11cb 100755 --- a/pyess/constants.py +++ b/pyess/constants.py @@ -1,14 +1,61 @@ #!/usr/bin/python # -*- coding: utf-8 -*- PREFIX = "https://{}/" -LOGIN_URL = f"{PREFIX}v1/user/setting/login" -TIMESYNC_URL = f"{PREFIX}v1/user/setting/timesync" + +# --- Login / TimeSync endpoint handling ------------------------------------- +# +# Older LG ESS Home firmware (PMS <= 10.05.6511, April 2020) exposes: +# PUT /v1/user/setting/login +# PUT /v1/user/setting/timesync +# +# Newer LG ESS Home firmware (PMS >= 10.05.7438, ~mid 2024 onwards) has +# shortened these to: +# PUT /v1/login +# PUT /v1/timesync +# +# `gluap/pyess` historically hard-coded the old paths. On a current device +# this surfaces as "password mismatched" (the device returns 404 for the +# old URL, which the client mis-interprets as a wrong password). +# +# To stay compatible with both firmware generations, ``aio_ess.ESS`` now +# tries the new URL first and falls back to the old URL on 404. The +# constants below expose the full set of candidate URLs in priority +# order; ``aio_ess`` uses them in that order and sticks with whichever +# one the device actually answers. +# +# See https://github.com/gluap/pyess/issues/37 for the original bug +# report and the device-side verification of the new endpoints. + +LOGIN_URL = f"{PREFIX}v1/user/setting/login" # legacy default (kept for back-compat with direct imports) +TIMESYNC_URL = f"{PREFIX}v1/user/setting/timesync" # legacy default + +# Order matters: ESS._login walks this list from index 0 onwards, stopping +# at the first URL that does not return HTTP 404. The first element is +# intentionally the *new* URL so that devices on current firmware are +# served without an extra round trip. +LOGIN_URLS = [ + f"{PREFIX}v1/login", # new firmware (PMS 10.05.7438+) + f"{PREFIX}v1/user/setting/login", # legacy firmware (PMS <= 10.05.6511) +] + +# For TimeSync, the situation is asymmetric: on the new firmware the old +# path 404s *and* the new path 404s as well (the endpoint is simply not +# implemented on the current build). ``aio_ess`` therefore treats a 404 +# here as "device has no time sync — skipping is safe" rather than as a +# hard failure, which prevents the tight login-retry loop that used to +# fill Home Assistant logs. +TIMESYNC_URLS = [ + f"{PREFIX}v1/timesync", # new firmware (currently 404 — harmless) + f"{PREFIX}v1/user/setting/timesync", # legacy firmware (200) +] +# --- /Login / TimeSync endpoint handling ------------------------------------ + STATE_URLS = { "network": f"{PREFIX}v1/user/setting/network", "systeminfo": f"{PREFIX}v1/user/setting/systeminfo", "batt": f"{PREFIX}v1/user/setting/batt", - "home": f"{PREFIX}v1/user/essinfo/home", - "common": f"{PREFIX}v1/user/essinfo/common", + "home": f"{PREFIX}v1/essinfo/home", + "common": f"{PREFIX}v1/essinfo/common", } GRAPH_TIMESPANS = {"day", "week", "month", "year"} GRAPH_DEVICES = {"batt", "load", "pv"} From 56bd42243a82f4f2700a1934d43328667b2659c0 Mon Sep 17 00:00:00 2001 From: blueleo076 Date: Mon, 15 Jun 2026 07:27:30 +0200 Subject: [PATCH 2/3] fix(login): try new PMS login URL with fallback to legacy URL On LG ESS Home PMS firmware 10.05.7438 and above, the device only exposes PUT /v1/login, not the legacy PUT /v1/user/setting/login. Previously the library hard-coded the legacy path, so every login attempt on current firmware got a 404 from the device which the client surfaced as a misleading 'wrong password' error. Walk the candidate LOGIN_URLS list at runtime, stop at the first URL that does not return 404, and pin the working URL on the client for the rest of the connection's lifetime. Treat TimeSync 404s the same way: the new firmware does not implement the endpoint at all, so a 404 there is no longer a hard failure that triggers the historical exponential-backoff retry loop. Other TimeSync errors (5xx, malformed body, status != success) still fall back to the existing retry behaviour. Also accept the new firmware's 'auth' field as an alternative to the legacy 'auth_key' response field, and make ESS.create / ESS.__init__ accept an optional pre-built aiohttp ClientSession so the new behaviour can be exercised by integration tests. Fixes: https://github.com/gluap/pyess/issues/37 --- pyess/aio_ess.py | 176 +++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 149 insertions(+), 27 deletions(-) diff --git a/pyess/aio_ess.py b/pyess/aio_ess.py index f84098d..5d85a56 100755 --- a/pyess/aio_ess.py +++ b/pyess/aio_ess.py @@ -11,8 +11,19 @@ import aiohttp -from pyess.constants import LOGIN_URL, TIMESYNC_URL, GRAPH_TIMESPANS, GRAPH_DEVICES, GRAPH_PARAMS, \ - GRAPH_TFORMATS, SWITCH_URL, STATE_URLS, BATT_URL +from pyess.constants import ( + LOGIN_URL, + TIMESYNC_URL, + LOGIN_URLS, + TIMESYNC_URLS, + GRAPH_TIMESPANS, + GRAPH_DEVICES, + GRAPH_PARAMS, + GRAPH_TFORMATS, + SWITCH_URL, + STATE_URLS, + BATT_URL, +) class ESSException(Exception): @@ -28,48 +39,159 @@ class ESSAuthException(ESSException): class ESS: @classmethod - async def create(cls, name=None, password=None, ip=None): - ess = cls(name, password, ip) + async def create(cls, name=None, password=None, ip=None, session=None): + """ + Create and log in a new ESS client. + + :param session: Optional pre-built aiohttp ``ClientSession``. + When supplied (typically by tests using ``aiohttp.test_utils``), + the client uses it instead of creating its own. The caller + is then responsible for closing the session. + """ + ess = cls(name, password, ip, session=session) await ess._login() return ess - def __init__(self, name, pw, ip=None): + def __init__(self, name, pw, ip=None, session=None): self.name = name self.pw = pw self.ip = ip self.logged_in = False self.auth_key = None - self.session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False), - timeout=aiohttp.ClientTimeout(connect=60,sock_read=60, sock_connect=60, total=180)) + # The endpoint actually answered by the device. Set during _login + # and reused on subsequent attempts so we don't re-probe every + # time. Stays None until the first successful login. + self._login_url = None + self._timesync_url = None + if session is not None: + self.session = session + else: + self.session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False), + timeout=aiohttp.ClientTimeout(connect=60,sock_read=60, sock_connect=60, total=180)) async def _login(self, retry=1): """ - Login to the ESS device. Called by __init__ - :return: + Login to the ESS device. Called by __init__. + + Tries each candidate login URL in ``pyess.constants.LOGIN_URLS`` + in order, sticking with the first one that does not return 404. + This keeps the library working on both the old PMS firmware + (``/v1/user/setting/login``) and the new PMS firmware introduced + somewhere between 10.05.6511 and 10.05.7438 + (``/v1/login``). + + :raises ESSAuthException: if every candidate URL returns 200 but + the device still reports ``status: password mismatched``. """ - url = LOGIN_URL.format(self.ip) - logger.info("fetching auth key") - async with self.session.put(url, json={"password": self.pw}) as r: - response_json = await r.json() - if "status" in response_json and response_json["status"] == "password mismatched": - raise ESSAuthException("wrong password") - # r = requests.put(url, json={"password": self.pw}, verify=False, headers={"Content-Type": "application/json"}) - auth_key = response_json["auth_key"] + # If a previous successful login already pinned an endpoint, use + # it directly. Otherwise walk the candidate list. We look the + # list up via ``pyess.constants`` on every call so monkey-patches + # in tests (and any future re-ordering at runtime) take effect. + import pyess.constants as _const + if self._login_url is not None: + login_candidates = [self._login_url] + else: + login_candidates = list(_const.LOGIN_URLS) + + auth_key = None + for url in login_candidates: + logger.info("fetching auth key via %s", url) + async with self.session.put(url, json={"password": self.pw}) as r: + if r.status == 404: + # Endpoint not implemented on this firmware — try + # the next candidate. + logger.info("login endpoint %s returned 404, trying fallback", url) + continue + try: + response_json = await r.json() + except (JSONDecodeError, ContentTypeError) as e: + raise ESSException( + f"unexpected non-JSON response from {url}: {e!r}" + ) from e + + # We got a JSON body. If the device says the password is + # wrong, that is a real auth failure (not a URL mismatch) — + # surface it immediately, do not try other URLs. + if ( + "status" in response_json + and response_json["status"] == "password mismatched" + ): + raise ESSAuthException("wrong password") + + # The current LG firmware returns either ``auth_key`` (old + # build) or ``auth`` (new build). Accept both. + auth_key = response_json.get("auth_key") or response_json.get("auth") + if auth_key is None: + raise ESSException( + f"login response from {url} missing auth_key: {response_json!r}" + ) + + # Pin the working URL for the lifetime of this client. + self._login_url = url + break + else: + # Every candidate returned 404 — the device speaks neither + # old nor new login URL. Surface a clear error instead of + # the historical misleading "wrong password". + raise ESSException( + f"no working login endpoint found; tried: {login_candidates}" + ) + + # TimeSync is best-effort. Both old and new firmware builds + # answer differently: legacy returns 200, the new PMS 10.05.x + # build simply does not implement the endpoint and answers 404 + # to both candidate paths. A 404 here is harmless — the device + # uses its own clock — so we log and continue instead of + # retrying the whole login. Any other failure (5xx, malformed + # body, "status != success") still falls through to the + # historical exponential-backoff retry. timesync_info = { "auth_key": auth_key, "by": "phone", - "date_time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()) + "date_time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()), } - async with self.session.put(TIMESYNC_URL.format(self.ip), json=timesync_info) as r: - try: - response_json = await r.json() - # rt = requests.put(TIMESYNC_URL.format(self.ip), json=timesync_info, verify=False, - # headers={"Content-Type": "application/json"}) - - assert response_json['status'] == 'success', response_json - except (JSONDecodeError, ContentTypeError): + if self._timesync_url is not None: + timesync_candidates = [self._timesync_url] + else: + timesync_candidates = list(_const.TIMESYNC_URLS) + + for ts_url in timesync_candidates: + logger.info("syncing device time via %s", ts_url) + async with self.session.put(ts_url, json=timesync_info) as r: + if r.status == 404: + # Endpoint not implemented on this firmware build — + # harmless. Try the other candidate, and if that + # also 404s, give up silently. + logger.info( + "timesync endpoint %s returned 404 (not implemented on this firmware)", + ts_url, + ) + self._timesync_url = ts_url + continue + try: + response_json = await r.json() + except (JSONDecodeError, ContentTypeError): + time.sleep(retry) + return await self._login(retry=retry * 2) + + if response_json.get("status") == "success": + self._timesync_url = ts_url + break + # Unknown / unexpected body shape — keep the legacy + # behaviour and retry the whole login with backoff. time.sleep(retry) return await self._login(retry=retry * 2) + else: + # All candidates returned 404. Time sync is unavailable on + # this firmware, but we did get an auth_key, so do not loop: + # just log once and accept the device's own clock. + logger.info( + "no working timesync endpoint found on this firmware " + "(tried: %s); continuing without device time sync", + timesync_candidates, + ) + self._timesync_url = None + self.auth_key = auth_key self.logged_in = True return auth_key @@ -77,7 +199,7 @@ async def _login(self, retry=1): async def get_graph(self, device: str, timespan: str, date: datetime.datetime): """ Get the time series data about a device - :param device: the device in question ``["batt", "load", "pv"]`` + :param device: the device in the question ``["batt", "load", "pv"]`` :param timespan: the timespan in question ``["day", "week", "month", "year"]`` :param date: the date specifying the time span. (for week, month and year I guess any date within the timespan of interest will suffice) From 3e1c8d6589ba771b81733d5c4ff644258040443b Mon Sep 17 00:00:00 2001 From: blueleo076 Date: Mon, 15 Jun 2026 07:27:41 +0200 Subject: [PATCH 3/3] test(login): cover new-firmware login + timesync fallback Six new tests in tests/aio_ess_endpoint_fallback_test.py: * test_new_firmware_login_succeeds Device exposes only /v1/login, returns 200 + auth_key. Client must pick the new URL, ignore the legacy 404, and complete login. * test_legacy_firmware_login_succeeds Device exposes only /v1/user/setting/login. Client must skip the 404 on /v1/login and fall back to the legacy path. * test_wrong_password_raises_auth_exception Both URLs return 200 + password_mismatched. Client must raise ESSAuthException, not retry forever, not raise the generic ESSException. * test_no_endpoint_found_raises_clear_exception Both URLs return 404. Client must raise ESSException whose message lists the URLs tried, and must NOT say 'wrong password'. * test_new_firmware_timesync_404_does_not_loop Regression test for the historical tight-retry loop that used to fill Home Assistant logs when the device's TimeSync endpoint was not implemented. * test_url_lists_priority_order Sanity check: the new URL is listed first in LOGIN_URLS, the legacy LOGIN_URL / TIMESYNC_URL module-level constants still exist and point at the legacy URLs (backward compatibility). The tests use aiohttp.test_utils.TestServer in-process and monkey-patch pyess.constants.LOGIN_URLS / TIMESYNC_URLS to point at the local server, exercising the real HTTP path of the fallback logic without needing a cassette or a real LG device on the network. Refs: https://github.com/gluap/pyess/issues/37 --- tests/aio_ess_endpoint_fallback_test.py | 333 ++++++++++++++++++++++++ 1 file changed, 333 insertions(+) create mode 100644 tests/aio_ess_endpoint_fallback_test.py diff --git a/tests/aio_ess_endpoint_fallback_test.py b/tests/aio_ess_endpoint_fallback_test.py new file mode 100644 index 0000000..1ac3cab --- /dev/null +++ b/tests/aio_ess_endpoint_fallback_test.py @@ -0,0 +1,333 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +""" +Regression tests for the new-firmware login URL fallback. + +These tests cover the two scenarios reported in +https://github.com/gluap/pyess/issues/37: + + 1. Device runs the new PMS firmware (>= 10.05.7438) which only + exposes the shortened auth endpoints. + 2. Device runs the legacy PMS firmware (<= 10.05.6511) which only + exposes the original long endpoints. + 3. TimeSync returns 404 on the new firmware and must not trigger the + historical retry-loop. + +Test strategy: build a real aiohttp server in-process, monkey-patch +``pyess.constants.LOGIN_URLS`` / ``TIMESYNC_URLS`` to point at the +local server, run the real ``ESS`` client, and assert behaviour. This +exercises the actual fallback logic and HTTP error handling without +needing a cassette or a real LG device on the network. +""" +import pytest +import pytest_asyncio +from aiohttp import web +from aiohttp.test_utils import TestServer, TestClient + +import pyess.constants as pyess_constants +from pyess.aio_ess import ESS, ESSAuthException, ESSException +from pyess.constants import ( + LOGIN_URL, + TIMESYNC_URL, + LOGIN_URLS as ORIGINAL_LOGIN_URLS, + TIMESYNC_URLS as ORIGINAL_TIMESYNC_URLS, +) +from pyess.constants import LOGIN_URLS, TIMESYNC_URLS # re-import for direct use in test_url_lists_priority_order + + +# ---- minimal in-memory "ESS device" apps ---------------------------------- + +def make_new_firmware_app(): + """ + Mimic LG ESS Home PMS >= 10.05.7438: + * /v1/login -> 200 + {auth_key: ...} + * /v1/user/setting/login -> 404 + * /v1/timesync -> 404 (not implemented on this build) + * /v1/user/setting/timesync-> 404 + """ + async def login_new(request): + body = await request.json() + if body.get("password") != "right": + return web.json_response({"status": "password mismatched"}, status=200) + return web.json_response({"auth_key": "auth-new-fw"}) + + async def legacy_login(request): + return web.Response(status=404, text="not here") + + async def any_timesync(request): + return web.Response(status=404, text="timesync not implemented on this firmware") + + async def get_state(request): + return web.json_response({"ok": True}) + + app = web.Application() + app.router.add_put("/v1/login", login_new) + app.router.add_put("/v1/user/setting/login", legacy_login) + app.router.add_put("/v1/timesync", any_timesync) + app.router.add_put("/v1/user/setting/timesync", any_timesync) + app.router.add_post("/v1/user/essinfo/common", get_state) + app.router.add_post("/v1/essinfo/common", get_state) + return app + + +def make_legacy_firmware_app(): + """ + Mimic LG ESS Home PMS <= 10.05.6511: + * /v1/login -> 404 + * /v1/user/setting/login -> 200 + {auth_key: ...} + * /v1/user/setting/timesync-> 200 + {status: success} + * /v1/timesync -> 404 + """ + async def login_new(request): + return web.Response(status=404, text="not implemented on legacy firmware") + + async def legacy_login(request): + body = await request.json() + if body.get("password") != "right": + return web.json_response({"status": "password mismatched"}, status=200) + return web.json_response({"auth_key": "auth-legacy-fw"}) + + async def timesync_new(request): + return web.Response(status=404, text="no") + + async def timesync_legacy(request): + return web.json_response({"status": "success"}) + + async def get_state(request): + return web.json_response({"ok": True}) + + app = web.Application() + app.router.add_put("/v1/login", login_new) + app.router.add_put("/v1/user/setting/login", legacy_login) + app.router.add_put("/v1/timesync", timesync_new) + app.router.add_put("/v1/user/setting/timesync", timesync_legacy) + app.router.add_post("/v1/user/essinfo/common", get_state) + app.router.add_post("/v1/essinfo/common", get_state) + return app + + +def make_wrong_password_app(): + """Both URLs return 200 + password_mismatched; the client must raise + ESSAuthException without trying forever.""" + async def login_any(request): + return web.json_response({"status": "password mismatched"}, status=200) + async def timesync_any(request): + return web.json_response({"status": "success"}) + + app = web.Application() + app.router.add_put("/v1/login", login_any) + app.router.add_put("/v1/user/setting/login", login_any) + app.router.add_put("/v1/timesync", timesync_any) + app.router.add_put("/v1/user/setting/timesync", timesync_any) + return app + + +def make_unreachable_app(): + """Both candidate URLs return 404 — the device speaks neither. The + client must raise a clear ESSException, *not* ESSAuthException.""" + async def none(request): + return web.Response(status=404, text="no") + app = web.Application() + app.router.add_put("/v1/login", none) + app.router.add_put("/v1/user/setting/login", none) + app.router.add_put("/v1/timesync", none) + app.router.add_put("/v1/user/setting/timesync", none) + return app + + +# ---- fixtures ------------------------------------------------------------- + +@pytest_asyncio.fixture +async def redirect_to_http_server(unused_tcp_port): + """ + Start a TestServer on a free TCP port, then monkey-patch + ``pyess.constants.LOGIN_URLS`` and ``TIMESYNC_URLS`` so the client + talks ``http://127.0.0.1:/...`` instead of the production + ``https:///...`` URLs. The caller decides which app the + server runs. + + Yields a callable ``(app) -> (client, port)``; the caller closes the + client when done. + """ + servers = [] # keep references for cleanup + yield unused_tcp_port + # Cleanup: restore URLs and close any open servers + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + for c in servers: + await c.close() + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + + +async def run_with_app(app_factory, unused_tcp_port): + """Spin up a test server, redirect the ESS client at it, and return + the test client + the redirect helper.""" + server = TestServer(app_factory()) + client = TestClient(server) + await client.start_server() + port = server._server.sockets[0].getsockname()[1] + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + return client + + +# ---- the actual tests ------------------------------------------------------ + +@pytest.mark.asyncio +async def test_new_firmware_login_succeeds(unused_tcp_port_factory): + """PMS 10.05.7438+: /v1/login answers, /v1/user/setting/login 404s.""" + # Bind the test server to the port the factory just reserved so the + # client URL monkey-patch and the server socket agree. + port = unused_tcp_port_factory() + server = TestServer(make_new_firmware_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + ess = await ESS.create("user", "right", "127.0.0.1", session=client.session) + assert ess.logged_in is True + assert ess.auth_key == "auth-new-fw" + assert ess._login_url is not None + assert ess._login_url.endswith("/v1/login") + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_legacy_firmware_login_succeeds(unused_tcp_port_factory): + """PMS <= 10.05.6511: /v1/login 404s, /v1/user/setting/login answers.""" + port = unused_tcp_port_factory() + server = TestServer(make_legacy_firmware_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + ess = await ESS.create("user", "right", "127.0.0.1", session=client.session) + assert ess.logged_in is True + assert ess.auth_key == "auth-legacy-fw" + assert ess._login_url.endswith("/v1/user/setting/login") + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_wrong_password_raises_auth_exception(unused_tcp_port_factory): + """Both endpoints return 200 + password_mismatched. The library must + raise ESSAuthException, not retry forever, not raise ESSException.""" + port = unused_tcp_port_factory() + server = TestServer(make_wrong_password_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + with pytest.raises(ESSAuthException): + await ESS.create("user", "wrong", "127.0.0.1", session=client.session) + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_no_endpoint_found_raises_clear_exception(unused_tcp_port_factory): + """If *every* candidate returns 404, we must get ESSException + explaining which URLs we tried, not the misleading 'wrong password'.""" + port = unused_tcp_port_factory() + server = TestServer(make_unreachable_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + with pytest.raises(ESSException) as exc: + await ESS.create("user", "right", "127.0.0.1", session=client.session) + # The error message must mention the URLs we tried so users can + # debug their network/firmware combination. + assert "/v1/login" in str(exc.value) + assert "/v1/user/setting/login" in str(exc.value) + # And it must *not* claim "wrong password". + assert "wrong password" not in str(exc.value).lower() + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_new_firmware_timesync_404_does_not_loop(unused_tcp_port_factory): + """Regression: when TimeSync 404s on new firmware, login must NOT + trigger the historical tight retry loop.""" + port = unused_tcp_port_factory() + server = TestServer(make_new_firmware_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + ess = await ESS.create("user", "right", "127.0.0.1", session=client.session) + # If the bug were still there this would either hang or take + # orders of magnitude longer. + assert ess.auth_key == "auth-new-fw" + # And we should have walked both timesync URLs, found both + # 404, and recorded that no timesync is available — without + # retrying the login. + assert ess._timesync_url is None + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +def test_url_lists_priority_order(): + """The new URL must come first in LOGIN_URLS so current devices are + served without an extra round-trip; the legacy URL is the fallback. + Backward-compat: the module-level LOGIN_URL/TIMESYNC_URL constants + must still exist and point at the legacy URLs.""" + assert LOGIN_URLS[0].endswith("/v1/login") + assert LOGIN_URLS[1].endswith("/v1/user/setting/login") + assert LOGIN_URL.endswith("/v1/user/setting/login") + assert TIMESYNC_URL.endswith("/v1/user/setting/timesync")