fix users endpoint (#278)

* fix users endpoint

* fix office filter to removes other office roles

* bump version 1.05

* mypy fix

---------

Co-authored-by: Eric Novotny <nov00002@umn.edu>

Author: Enovotny

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

@@ -35,18 +35,66 @@ def get_user_profile() -> dict[str, Any]:
     return dict(response)
 
 
+def filter_users_by_office(data: dict[str, Any], office: str) -> dict[str, Any]:
+    """
+    Filter users JSON to only include users that have roles for the specified office.
+    Each user's roles dict will only contain the entry for that office.
+
+    Args:
+        data:   The full users JSON as a Python dict.
+        office: The office key to filter by (e.g., 'MVP', 'LRL').
+
+    Returns:
+        A new dict with the same structure, filtered to the specified office.
+    """
+    filtered_users = []
+
+    for user in data.get("users", []):
+        roles = user.get("roles", {})
+
+        if office in roles:
+            # Build a copy of the user with only the target office's roles
+            filtered_user = {k: v for k, v in user.items() if k != "roles"}
+            filtered_user["roles"] = {office: roles[office]}
+            filtered_users.append(filtered_user)
+
+    return {
+        "page": data.get("page"),
+        "page-size": data.get("page-size"),
+        "total": len(filtered_users),
+        "users": filtered_users,
+    }
+
+
 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:
+        response = filter_users_by_office(response, office_id)
     return Data(response, selector="users")
 
 

pyproject.toml

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

tests/cda/users/users_CDA_test.py

@@ -107,6 +107,53 @@ 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).json
+    users_list = users.get("users", [])
+    assert isinstance(users_list, list)
+    assert any(
+        str(u.get("user-name", "")).lower() == user_name.lower() for u in users_list
+    )
+    for user in users["users"]:
+        roles = user["roles"]
+        assert set(roles.keys()) == {
+            office_id
+        }, f"{user['user-name']} has unexpected office keys: {set(roles.keys())}"
+
+
+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