fix users endpoint

cwms/catalog/catalog.py

@@ -67,7 +67,9 @@ def get_locations_catalog(
         "location-kind-like": location_kind_like,
     }
 
-    response = api.get(endpoint=endpoint, params=params, api_version=2)
+    response = api.get_with_paging(
+        endpoint=endpoint, selector="entries", params=params, api_version=2
+    )
     return Data(response, selector="entries")
 
 
@@ -131,7 +133,9 @@ def get_timeseries_catalog(
         "include-extents": include_extents,
     }
 
-    response = api.get(endpoint=endpoint, params=params, api_version=2)
+    response = api.get_with_paging(
+        endpoint=endpoint, selector="entries", params=params, api_version=2
+    )
     return Data(response, selector="entries")
 
 

cwms/users/users.py

@@ -37,16 +37,37 @@ def get_user_profile() -> dict[str, Any]:
 
 def get_users(
     office_id: Optional[str] = None,
+    username_like: Optional[str] = None,
+    include_roles: Optional[bool] = None,
     page: Optional[str] = None,
-    page_size: Optional[int] = None,
+    page_size: Optional[int] = 5000,
 ) -> Data:
     """Retrieve users with optional office and paging filters."""
 
-    params = {"office": office_id, "page": page, "page-size": page_size}
+    endpoint = "users"
+    params = {
+        "office": office_id,
+        "username-like": username_like,
+        "include-roles": include_roles,
+        "page": page,
+        "page-size": page_size,
+    }
     try:
-        response = api.get("users", params=params, api_version=1)
+        response = api.get_with_paging(
+            endpoint=endpoint, selector="users", params=params, api_version=1
+        )
     except api.ApiError as error:
         _raise_user_management_error(error, "User list lookup")
+
+    # filter by office if office_id is provided since the API does not
+    # currently support filtering by office on the backend. This is a
+    # temporary workaround until the API supports office filtering.
+    if office_id:
+        data = response
+        filtered_users = [
+            user for user in data["users"] if office_id in user.get("roles", {})
+        ]
+        response = {**data, "users": filtered_users, "total": len(filtered_users)}
     return Data(response, selector="users")
 
 

pyproject.toml

@@ -2,7 +2,7 @@
 name = "cwms-python"
 repository = "https://github.com/HydrologicEngineeringCenter/cwms-python"
 
-version = "1.0.3"
+version = "1.0.4"
 
 
 packages = [

tests/cda/users/users_CDA_test.py

@@ -107,6 +107,48 @@ def test_get_users():
     )
 
 
+def test_get_users_by_office():
+    # /users?office={office_id}
+    # Covers filtering the user list by office and verifies the known test account appears in the
+    # returned page when filtering by the test user's office.
+    user_name = str(TEST_USER_NAME)
+    office_id = TEST_OFFICE_ID
+    users = cwms.get_users(office_id=office_id, page_size=200)
+    users_list = users.json.get("users", [])
+    assert isinstance(users_list, list)
+    assert any(
+        str(u.get("user-name", "")).lower() == user_name.lower() for u in users_list
+    )
+
+
+def test_get_users_by_office_not_present_in_other_office():
+    # /users?office={office_id}
+    # Covers filtering the user list by office and verifies the known test account does not appear in the
+    # returned page when filtering by a different office.
+    user_name = str(TEST_USER_NAME)
+    office_id = "MVP" if TEST_OFFICE_ID != "MVP" else "SPK"
+    users = cwms.get_users(office_id=office_id, page_size=200)
+    users_list = users.json.get("users", [])
+    assert isinstance(users_list, list)
+    assert not any(
+        str(u.get("user-name", "")).lower() == user_name.lower() for u in users_list
+    )
+
+
+def test_get_users_with_paging():
+    # /users with paging parameters
+    # Covers requesting a specific page and page size and verifies the response contains the expected pagination metadata.
+    page_size = 2
+    users_with_page = cwms.get_users(page_size=page_size)
+    users_list = users_with_page.json.get("users", [])
+    users_without_page = cwms.get_users()
+    all_users_list = users_without_page.json.get("users", [])
+    assert isinstance(users_list, list)
+    assert isinstance(all_users_list, list)
+    assert len(users_list) >= page_size
+    assert len(all_users_list) == len(users_list)
+
+
 def test_store_update_delete_user_roles_roundtrip():
     # Round Trip: Covers the role-management lifecycle for an existing user in one office:
     # 1. add a role the user does not currently have