Add Lafayette Utilities System (LUS)

README.md

@@ -59,6 +59,7 @@ series is available.
   - PECO Energy Company (PECO)
   - Potomac Electric Power Company (Pepco)
 - Glendale Water and Power (GWP)
+- Lafayette Utilities System (LUS)
 - Northern Indiana Public Service Company (NIPSCO)
 - Pacific Gas & Electric (PG&E)
 - Puget Sound Energy (PSE)

src/opower/utilities/lus.py

@@ -0,0 +1,73 @@
+"""Lafayette Utilities System (LUS)."""
+
+from typing import Any
+
+import aiohttp
+
+from ..const import USER_AGENT
+from ..exceptions import InvalidAuth
+from .base import UtilityBase
+
+
+class LUS(UtilityBase):
+    """Lafayette Utilities System (LUS).
+
+    LUS, the municipal utility of Lafayette, Louisiana, uses the Opower Digital
+    Self-Service (DSS) portal at ``dss-lus.opower.com`` for electric, water and
+    wastewater. Credentials are posted to the portal's login API, which answers
+    with a ``sessionToken`` used as the bearer token for later API calls.
+    """
+
+    @staticmethod
+    def name() -> str:
+        """Distinct recognizable name of the utility."""
+        return "Lafayette Utilities System (LUS)"
+
+    def subdomain(self) -> str:
+        """Return the opower.com subdomain for this utility."""
+        return "lus"
+
+    @staticmethod
+    def timezone() -> str:
+        """Return the timezone.
+
+        Should match the siteTimeZoneId of the API responses.
+        """
+        return "America/Chicago"
+
+    @staticmethod
+    def is_dss() -> bool:
+        """Check if Utility using DSS version of the portal."""
+        return True
+
+    async def async_login(
+        self,
+        session: aiohttp.ClientSession,
+        username: str,
+        password: str,
+        login_data: dict[str, Any],
+    ) -> str | None:
+        """Login to the utility website and return the session token."""
+        session.cookie_jar.clear(lambda cookie: cookie["domain"].endswith(".opower.com"))
+
+        async with session.post(
+            "https://dss-lus.opower.com/webcenter/edge/apis/identity-management-v1/cws/v1/lus/login",
+            data={"username": username, "password": password, "rememberMe": "true"},
+            headers={"User-Agent": USER_AGENT},
+        ) as response:
+            if response.status in (400, 401, 403):
+                raise InvalidAuth("Username/Password are invalid")
+            response.raise_for_status()
+            try:
+                result = await response.json()
+            except (aiohttp.ContentTypeError, ValueError) as err:
+                # Not retryable: most likely the login API changed and needs a code change
+                raise InvalidAuth("Unexpected response from the LUS login") from err
+
+        token = result.get("sessionToken") if isinstance(result, dict) else None
+        if not token:
+            raise InvalidAuth("Login response had no sessionToken")
+
+        # Selects the DSS version of the portal for later requests
+        session.cookie_jar.update_cookies({"dssPortalCW": "1"})
+        return str(token)

tests/opower/utilities/test_lus.py

@@ -0,0 +1,120 @@
+"""Tests for Lafayette Utilities System (LUS)."""
+
+import unittest
+from typing import Any
+
+from opower import get_supported_utility_names, select_utility
+from opower.exceptions import InvalidAuth
+from opower.utilities.lus import LUS
+
+
+class _FakeResponse:
+    """Minimal stand-in for an aiohttp response."""
+
+    def __init__(self, status: int, body: Any) -> None:
+        self.status = status
+        self._body = body
+
+    async def json(self) -> Any:
+        if isinstance(self._body, Exception):
+            raise self._body
+        return self._body
+
+    def raise_for_status(self) -> None:
+        if self.status >= 400:
+            raise AssertionError(f"unexpected raise_for_status for {self.status}")
+
+    async def __aenter__(self) -> "_FakeResponse":
+        return self
+
+    async def __aexit__(self, *args: object) -> bool:
+        return False
+
+
+class _FakeCookieJar:
+    def __init__(self) -> None:
+        self.cookies: dict[str, str] = {}
+
+    def clear(self, predicate: Any = None) -> None:
+        self.cookies.clear()
+
+    def update_cookies(self, cookies: dict[str, str]) -> None:
+        self.cookies.update(cookies)
+
+
+class _FakeSession:
+    """Session that answers the login POST with a fixed status and body."""
+
+    def __init__(self, status: int, body: Any) -> None:
+        self._status = status
+        self._body = body
+        self.cookie_jar = _FakeCookieJar()
+        self.posts: list[tuple[str, dict[str, Any]]] = []
+
+    def post(self, url: str, **kwargs: Any) -> _FakeResponse:
+        self.posts.append((url, kwargs))
+        return _FakeResponse(self._status, self._body)
+
+
+class TestLUS(unittest.IsolatedAsyncioTestCase):
+    """Test Lafayette Utilities System (LUS)."""
+
+    def test_name(self) -> None:
+        """Test name."""
+        self.assertEqual("Lafayette Utilities System (LUS)", LUS().name())
+
+    def test_subdomain(self) -> None:
+        """Test subdomain."""
+        self.assertEqual("lus", LUS().subdomain())
+
+    def test_timezone(self) -> None:
+        """Test timezone."""
+        self.assertEqual("America/Chicago", LUS().timezone())
+
+    def test_is_dss(self) -> None:
+        """LUS uses the DSS portal at dss-lus.opower.com."""
+        self.assertTrue(LUS().is_dss())
+
+    def test_selectable_by_name(self) -> None:
+        """Test the utility is listed and can be selected."""
+        self.assertIn("Lafayette Utilities System (LUS)", get_supported_utility_names())
+        self.assertIs(select_utility("Lafayette Utilities System (LUS)"), LUS)
+
+    async def test_login_returns_session_token(self) -> None:
+        """The login API's sessionToken is the bearer token."""
+        session = _FakeSession(200, {"sessionToken": "abc-123"})
+
+        token = await LUS().async_login(session, "user", "pw", {})  # type: ignore[arg-type]
+
+        self.assertEqual("abc-123", token)
+        url, kwargs = session.posts[0]
+        self.assertEqual(
+            "https://dss-lus.opower.com/webcenter/edge/apis/identity-management-v1/cws/v1/lus/login",
+            url,
+        )
+        self.assertEqual({"username": "user", "password": "pw", "rememberMe": "true"}, kwargs["data"])
+        self.assertEqual({"dssPortalCW": "1"}, session.cookie_jar.cookies)
+
+    async def test_rejected_credentials_raise_invalid_auth(self) -> None:
+        """A 401 is a credential rejection."""
+        session = _FakeSession(401, {"error": {"serviceErrorCode": "INVALID_CREDENTIALS"}})
+        with self.assertRaises(InvalidAuth):
+            await LUS().async_login(session, "user", "bad", {})  # type: ignore[arg-type]
+
+    async def test_missing_token_raises_invalid_auth(self) -> None:
+        """A 200 without a sessionToken is not a usable login."""
+        session = _FakeSession(200, {})
+        with self.assertRaises(InvalidAuth):
+            await LUS().async_login(session, "user", "pw", {})  # type: ignore[arg-type]
+
+    async def test_non_json_reply_raises_invalid_auth(self) -> None:
+        """A non-JSON reply most likely means the API changed, so it shouldn't be retried."""
+        session = _FakeSession(200, ValueError("Expecting value"))
+        with self.assertRaises(InvalidAuth):
+            await LUS().async_login(session, "user", "pw", {})  # type: ignore[arg-type]
+
+    async def test_non_object_json_raises_invalid_auth(self) -> None:
+        """A JSON reply that isn't an object has no sessionToken."""
+        session = _FakeSession(200, ["unexpected"])
+        with self.assertRaises(InvalidAuth):
+            await LUS().async_login(session, "user", "pw", {})  # type: ignore[arg-type]