diff --git a/cwms/locations/lookups.py b/cwms/locations/lookups.py new file mode 100644 index 00000000..7e8c2089 --- /dev/null +++ b/cwms/locations/lookups.py @@ -0,0 +1,162 @@ +# Copyright (c) 2026 +# United States Army Corps of Engineers - Hydrologic Engineering Center (USACE/HEC) +# All Rights Reserved. USACE PROPRIETARY/CONFIDENTIAL. +# Source may not be released without written approval from HEC + +import cwms.api as api +from cwms.cwms_types import JSON, Data + +ENDPOINT = "lookup-types" + + +def get_all_lookups(category: str, prefix: str, office_id: str) -> Data: + """ + Retrieves all lookups for a given category, prefix, and office. + + Parameters + ---------- + category : str + Filters lookup types to the specified category + prefix : str + Filters lookup types to the specified prefix + office_id : str + Filters lookup types to the specified office ID + + Returns + ------- + Data + The JSON response from CWMS Data API wrapped in a Data object. + + Raises + ------ + ValueError + If any required argument is missing. + ClientError + If a 400 range error code response is returned from the server. + NoDataFoundError + If a 404 range error code response is returned from the server. + ServerError + If a 500 range error code response is returned from the server. + """ + if not all([category, prefix, office_id]): + raise ValueError("Category, Prefix, and Office ID must be specified") + + params = {"category": category, "prefix": prefix, "office": office_id} + response = api.get(ENDPOINT, params, api_version=1) + return Data(response) + + +def create_lookup(data: JSON, category: str, prefix: str) -> None: + """ + Creates a new lookup entry. + + Parameters + ---------- + data: JSON + A dictionary representing the JSON data to be stored. This should match the + LookupType structure as defined by the API. + category : str + Specifies the category of the lookup. + prefix : str + Specifies the prefix of the lookup. + + Returns + ------- + None + + Raises + ------ + ValueError + If any required argument is missing. + ClientError + If a 400 range error code response is returned from the server. + NoDataFoundError + If a 404 range error code response is returned from the server. + ServerError + If a 500 range error code response is returned from the server. + """ + if not all([category, prefix]): + raise ValueError("Category and Prefix must be specified") + if not data: + raise ValueError("Data must be specified") + params = {"category": category, "prefix": prefix} + api.post(ENDPOINT, data, params, api_version=1) + + +def update_lookup(data: JSON, name: str, category: str, prefix: str) -> None: + """ + Updates a specified lookup entry. + + Parameters + ---------- + data : JSON + A dictionary representing the JSON data to be stored. + If the `data` value is None, a `ValueError` will be raised. + name : str + Specifies the location type to update + category : str + Specifies the category of the lookup. + prefix : str + Specifies the prefix of the lookup. + + Returns + ------- + None + + Raises + ------ + ValueError + If any required argument is missing. + ClientError + If a 400 range error code response is returned from the server. + NoDataFoundError + If a 404 range error code response is returned from the server. + ServerError + If a 500 range error code response is returned from the server. + """ + if not all([name, category, prefix]): + raise ValueError("Name, Category, and Prefix must be specified") + if not data: + raise ValueError("Data must be specified") + + endpoint = f"{ENDPOINT}/{name}" + params = {"category": category, "prefix": prefix} + api.patch(endpoint, data, params, api_version=1) + + +def delete_lookup(name: str, category: str, prefix: str, office_id: str) -> None: + """ + Deletes a specified lookup entry. + + Parameters + ---------- + name : str + Specifies the location type to delete. + category : str + Specifies the category id of the lookup type to be deleted. + prefix : str + Specifies the prefix of the lookup type to be deleted. + office_id : str + Specifies the owning office of the lookup type to be deleted. + + Returns + ------- + None + + Raises + ------ + ValueError + If any required argument is missing. + ClientError + If a 400 range error code response is returned from the server. + NoDataFoundError + If a 404 range error code response is returned from the server. + ServerError + If a 500 range error code response is returned from the server. + """ + if not all([name, category, prefix, office_id]): + raise ValueError("Name, Category, Prefix, and Office ID must be specified") + + endpoint = f"{ENDPOINT}/{name}" + params = {"category": category, "prefix": prefix, "office": office_id} + api.delete(endpoint, params, api_version=1) diff --git a/tests/cda/locations/lookup_test.py b/tests/cda/locations/lookup_test.py new file mode 100644 index 00000000..17f5b57d --- /dev/null +++ b/tests/cda/locations/lookup_test.py @@ -0,0 +1,127 @@ +import pytest + +import cwms.api as api +import cwms.locations.lookups as lookups +from cwms.cwms_types import Data + +OFFICE_ID = "SPK" +CATEGORY = "USACE" +PREFIX = "L" +DISPLAY_VALUE = "Test Lookup" +TOOLTIP = "Test Tooltip" + +LOOKUP_DATA = { + "office-id": OFFICE_ID, + "display-value": DISPLAY_VALUE, + "tooltip": TOOLTIP, + "active": True, +} + + +def _cleanup(): + try: + lookups.delete_lookup(DISPLAY_VALUE, CATEGORY, PREFIX, OFFICE_ID) + except Exception: + pass + + +@pytest.fixture(scope="module", autouse=True) +def setup_data(): + _cleanup() + lookups.create_lookup(LOOKUP_DATA, CATEGORY, PREFIX) + + +@pytest.fixture(autouse=True) +def init_session(): + print("Initializing CWMS API session for lookup type tests...") + + +def test_get_all_lookups(): + new_display_value = DISPLAY_VALUE + " Refreshed" + data = { + "office-id": OFFICE_ID, + "display-value": new_display_value, + "tooltip": TOOLTIP, + "active": True, + } + lookups.create_lookup(data, CATEGORY, PREFIX) + + result = lookups.get_all_lookups(CATEGORY, PREFIX, OFFICE_ID) + found = False + found2 = False + for item in result.json: + if item.get("display-value") == new_display_value: + found = True + if item.get("display-value") == DISPLAY_VALUE: + found2 = True + assert ( + found + ), f"Lookup with display-value {new_display_value} not found after creation" + assert found2, f"Lookup with display-value {DISPLAY_VALUE} not found after creation" + + +def test_create_lookup(): + display_value = DISPLAY_VALUE + " Created" + data = { + "office-id": OFFICE_ID, + "display-value": display_value, + "tooltip": TOOLTIP, + "active": True, + } + lookups.create_lookup(data, CATEGORY, PREFIX) + + result = lookups.get_all_lookups(CATEGORY, PREFIX, OFFICE_ID) + found = False + for item in result.json: + if item.get("display-value") == display_value: + found = True + break + assert found, f"Lookup with display-value {display_value} not found after creation" + + +def test_update_lookup(): + new_display_value = DISPLAY_VALUE + " Updated" + data = { + "office-id": OFFICE_ID, + "display-value": new_display_value, + "tooltip": "Updated Tooltip", + "active": True, + } + # The name parameter in update_lookup corresponds to the display-value of the item to update + lookups.update_lookup(data, DISPLAY_VALUE, CATEGORY, PREFIX) + + result = lookups.get_all_lookups(CATEGORY, PREFIX, OFFICE_ID) + found = False + for item in result.json: + if item.get("display-value") == new_display_value: + found = True + break + assert found, f"Lookup with updated display-value {new_display_value} not found" + + try: + lookups.delete_lookup(new_display_value, CATEGORY, PREFIX, OFFICE_ID) + except Exception: + pass + + +def test_delete_lookup(): + office = "LRL" + data = { + "office-id": office, + "display-value": DISPLAY_VALUE, + "tooltip": TOOLTIP, + "active": True, + } + lookups.create_lookup(data, CATEGORY, PREFIX) + + lookups.delete_lookup(DISPLAY_VALUE, CATEGORY, PREFIX, office) + + result = lookups.get_all_lookups(CATEGORY, PREFIX, office) + found = False + for item in result.json: + if item.get("display-value") == DISPLAY_VALUE: + found = True + break + assert ( + not found + ), f"Lookup with display-value {DISPLAY_VALUE} still found after deletion"