fix(cm8200): persist status cache across poll cycle, add re-auth validation (#168)

The CM8200 driver cached the status page during login() but consumed it
on the first _fetch_status_page() call. In the real collector flow
(login -> get_device_info -> get_docsis_data), get_device_info consumed
the cache, forcing get_docsis_data to trigger a full re-authentication.
This is problematic because the CM8200A is sensitive to repeated auth
attempts and can enter a 5-minute brute-force lockout.

Changes:
- Keep status cache for the entire poll cycle instead of consuming it
- Invalidate cache at the start of login() so each poll gets fresh data
- Add token and login-page validation to _fetch_status_page() re-auth
  path (matching the existing checks in login())
- Cache successful re-auth results to avoid further round-trips
- Add tests covering the real collector flow (single auth for full poll)
  and re-auth validation (token rejection, login page detection)

Co-authored-by: itsDNNS <itsDNNS@users.noreply.github.com>

Author: itsDNNS

app/drivers/cm8200.py

@@ -91,6 +91,7 @@ def login(self) -> None:
         error. It returns the login page (HTTP 200, ~4170 bytes) instead
         of the status page when locked out.
         """
+        self._status_html = None
         creds = base64.b64encode(f"{self._user}:{self._password}".encode()).decode()
         for attempt in range(2):
             try:
@@ -189,14 +190,17 @@ def get_connection_info(self) -> dict:
     def _fetch_status_page(self) -> BeautifulSoup:
         """Fetch and parse the status page HTML.
 
-        Reuses cached HTML from login if available (same page).
-        Re-authenticates when the cache has been consumed.
+        Reuses cached HTML from login for the entire poll cycle.  The
+        cache is refreshed on the next ``login()`` call, so both
+        ``get_device_info()`` and ``get_docsis_data()`` share the same
+        snapshot without triggering a second auth round-trip.
+
+        If no cache is available (e.g. session expired mid-poll), falls
+        back to a full re-auth with the same validation as ``login()``.
         """
         if self._status_html:
-            html = self._status_html
-            self._status_html = None
-            log.debug("CM8200 using cached status HTML (%d bytes)", len(html))
-            return BeautifulSoup(html, "html.parser")
+            log.debug("CM8200 using cached status HTML (%d bytes)", len(self._status_html))
+            return BeautifulSoup(self._status_html, "html.parser")
 
         creds = base64.b64encode(f"{self._user}:{self._password}".encode()).decode()
         try:
@@ -205,6 +209,14 @@ def _fetch_status_page(self) -> BeautifulSoup:
                 timeout=30,
             )
             token = r1.text.strip()
+
+            if len(token) > 64 or not token.isalnum():
+                raise RuntimeError(
+                    "CM8200 re-auth failed: expected session token but received "
+                    f"{len(token)} bytes (wrong credentials or brute-force "
+                    "lockout, wait 5 minutes)"
+                )
+
             self._cookie_header = f"HttpOnly: true, Secure: true; credential={token}"
 
             r = self._session.get(
@@ -213,8 +225,17 @@ def _fetch_status_page(self) -> BeautifulSoup:
                 timeout=30,
             )
             r.raise_for_status()
+
+            if len(r.text) < 5000 or "downstream bonded" not in r.text.lower():
+                raise RuntimeError(
+                    "CM8200 re-auth succeeded but status page not returned "
+                    f"({len(r.text)} bytes). Modem may be in brute-force "
+                    "lockout (wait 5 minutes)."
+                )
         except requests.RequestException as e:
             raise RuntimeError(f"CM8200 status page retrieval failed: {e}")
+
+        self._status_html = r.text
         log.debug("CM8200 status page fetched (%d bytes)", len(r.text))
         return BeautifulSoup(r.text, "html.parser")
 

tests/test_cm8200_driver.py

@@ -462,12 +462,104 @@ def test_missing_tr_on_column_headers(self, driver):
             assert len(data["channelUs"]["docsis30"]) == 1
             assert data["channelUs"]["docsis30"][0]["channelID"] == 3
 
-    def test_status_html_cache_consumed(self, driver):
-        """Cached HTML from login is consumed on first fetch, then cleared."""
+    def test_status_html_cache_persists_across_calls(self, driver):
+        """Cached HTML from login is reused across multiple fetches."""
         driver._status_html = SAMPLE_STATUS_HTML
-        soup = driver._fetch_status_page()
+        soup1 = driver._fetch_status_page()
+        assert soup1.find("span", id="thisModelNumberIs") is not None
+        assert driver._status_html is not None
+        soup2 = driver._fetch_status_page()
+        assert soup2.find("span", id="thisModelNumberIs") is not None
+
+    def test_login_clears_stale_cache(self, driver):
+        """login() invalidates old cache before authenticating."""
+        driver._status_html = "<html>stale</html>"
+
+        token_response = MagicMock()
+        token_response.raise_for_status = MagicMock()
+        token_response.text = "freshtoken123"
+
+        status_response = MagicMock()
+        status_response.raise_for_status = MagicMock()
+        status_response.text = SAMPLE_STATUS_HTML
+
+        with patch.object(driver._session, "get", side_effect=[token_response, status_response]):
+            driver.login()
+
+        assert driver._status_html == SAMPLE_STATUS_HTML
+
+
+# -- Collector flow (login -> device_info -> docsis_data) --
+
+class TestCollectorFlow:
+    def test_single_auth_for_full_poll(self, driver):
+        """Real collector calls login(), get_device_info(), get_docsis_data().
+
+        All three must work with a single auth round-trip.  get_device_info()
+        must not consume the cache so get_docsis_data() can reuse it.
+        """
+        token_response = MagicMock()
+        token_response.raise_for_status = MagicMock()
+        token_response.text = "token123"
+
+        status_response = MagicMock()
+        status_response.raise_for_status = MagicMock()
+        status_response.text = SAMPLE_STATUS_HTML
+
+        with patch.object(driver._session, "get", side_effect=[token_response, status_response]) as mock_get:
+            driver.login()
+            info = driver.get_device_info()
+            data = driver.get_docsis_data()
+
+        # Only 2 GETs: auth token + status page (both during login)
+        assert mock_get.call_count == 2
+        assert info["model"] == "CM8200A"
+        assert len(data["channelDs"]["docsis30"]) == 32
+        assert len(data["channelDs"]["docsis31"]) == 1
+
+    def test_fetch_reauth_validates_token(self, driver):
+        """_fetch_status_page re-auth rejects HTML instead of token."""
+        driver._status_html = None
+        driver._cookie_header = None
+
+        html_response = MagicMock()
+        html_response.text = "<html><body>Login page</body></html>"
+
+        with patch.object(driver._session, "get", return_value=html_response):
+            with pytest.raises(RuntimeError, match="re-auth failed"):
+                driver._fetch_status_page()
+
+    def test_fetch_reauth_validates_status_page(self, driver):
+        """_fetch_status_page re-auth rejects small/login pages."""
+        driver._status_html = None
+
+        token_response = MagicMock()
+        token_response.text = "validtoken99"
+
+        login_page = MagicMock()
+        login_page.raise_for_status = MagicMock()
+        login_page.text = "<html>small login page</html>"
+
+        with patch.object(driver._session, "get", side_effect=[token_response, login_page]):
+            with pytest.raises(RuntimeError, match="re-auth succeeded but status page not returned"):
+                driver._fetch_status_page()
+
+    def test_fetch_reauth_caches_result(self, driver):
+        """Successful re-auth in _fetch_status_page caches the HTML."""
+        driver._status_html = None
+
+        token_response = MagicMock()
+        token_response.text = "reauth123"
+
+        status_response = MagicMock()
+        status_response.raise_for_status = MagicMock()
+        status_response.text = SAMPLE_STATUS_HTML
+
+        with patch.object(driver._session, "get", side_effect=[token_response, status_response]):
+            soup = driver._fetch_status_page()
+
+        assert driver._status_html == SAMPLE_STATUS_HTML
         assert soup.find("span", id="thisModelNumberIs") is not None
-        assert driver._status_html is None
 
 
 # -- Analyzer integration --