Initial implementation for properties endpoint support

cwms/__init__.py

@@ -18,6 +18,7 @@
 from cwms.projects.project_locks import *
 from cwms.projects.projects import *
 from cwms.projects.water_supply.accounting import *
+from cwms.properties.properties import *
 from cwms.ratings.ratings import *
 from cwms.ratings.ratings_spec import *
 from cwms.ratings.ratings_template import *

cwms/properties/properties.py

@@ -0,0 +1,168 @@
+#  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
+
+from typing import Optional
+
+import cwms.api as api
+from cwms.cwms_types import JSON, Data
+
+
+def get_properties(
+    office_mask: Optional[str] = None,
+    category_id_mask: Optional[str] = None,
+    name_mask: Optional[str] = None,
+) -> Data:
+    """
+    Returns matching CWMS Property Data.
+
+    Parameters
+    ----------
+        office_mask: string, optional
+            Filters properties to the specified office mask
+        category_id_mask: string, optional
+            Filters properties to the specified category mask
+        name_mask: string, optional
+            Filters properties to the specified name mask
+
+    Returns
+    -------
+        cwms data type
+    """
+
+    endpoint = "properties"
+    params = {
+        "office-mask": office_mask,
+        "category-id-mask": category_id_mask,
+        "name-mask": name_mask,
+    }
+
+    response = api.get(endpoint, params, api_version=1)
+    return Data(response)
+
+
+def get_property(
+    name: str, office: str, category_id: str, default_value: Optional[str] = None
+) -> Data:
+    """
+    Returns CWMS Property Data.
+
+    Parameters
+    ----------
+        name: string
+            Specifies the name of the property to be retrieved.
+        office: string
+            Specifies the owning office of the property to be retrieved.
+        category_id: string
+            Specifies the category id of the property to be retrieved.
+        default_value: string, optional
+            Specifies the default value if the property does not exist.
+
+    Returns
+    -------
+        cwms data type
+    """
+
+    endpoint = f"properties/{name}"
+    params = {
+        "office": office,
+        "category-id": category_id,
+        "default-value": default_value,
+    }
+
+    response = api.get(endpoint, params, api_version=1)
+    return Data(response)
+
+
+def create_property(data: JSON) -> None:
+    """
+    Create CWMS Property.
+
+    Parameters
+    ----------
+        data: JSON dictionary
+            Property data to be stored.
+            Example:
+            {
+                "office-id": "string",
+                "name": "string",
+                "category": "string",
+                "value": "string,
+                "comment": "string
+            }
+
+    Returns
+    -------
+        None
+    """
+
+    endpoint = "properties"
+
+    if data is None:
+        raise ValueError("Cannot store a property without JSON data")
+
+    return api.post(endpoint, data, api_version=1)
+
+
+def update_property(name: str, data: JSON) -> None:
+    """
+    Update CWMS Property.
+
+    Parameters
+    ----------
+        name: string
+            Specifies the name of the property to be updated.
+        data: JSON dictionary
+            Property data to be updated.
+            Example:
+            {
+                "office-id": "string",
+                "name": "string",
+                "category": "string",
+                "value": "string",
+                "comment": "string
+            }
+
+    Returns
+    -------
+        None
+    """
+
+    endpoint = f"properties/{name}"
+
+    if name is None:
+        raise ValueError("Must specify a property name to update")
+
+    if data is None:
+        raise ValueError("Cannot update a property without JSON data")
+
+    return api.patch(endpoint, data, api_version=1)
+
+
+def delete_property(name: str, office: str, category_id: str) -> None:
+    """
+    Delete CWMS Property.
+
+    Parameters
+    ----------
+        name: string
+            Specifies the name of the property to be deleted.
+        office: string
+            Specifies the owning office of the property to be deleted.
+        category_id: string
+            Specifies the category id of the property to be deleted.
+
+    Returns
+    -------
+        None
+    """
+
+    endpoint = f"properties/{name}"
+
+    params = {
+        "office": office,
+        "category-id": category_id,
+    }
+
+    return api.delete(endpoint, params, api_version=1)

tests/cda/properties/properties_CDA_test.py

@@ -0,0 +1,110 @@
+#  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 pytest
+
+import cwms.api as api
+import cwms.properties.properties as properties
+from cwms.api import ApiError
+
+TEST_OFFICE = "SPK"
+TEST_CATEGORY = "PytestCategory"
+TEST_NAME = "PytestProperty"
+TEST_VALUE = "PytestValue"
+TEST_COMMENT = "PytestComment"
+
+TEST_PROPERTY_DATA = {
+    "office-id": TEST_OFFICE,
+    "name": TEST_NAME,
+    "category": TEST_CATEGORY,
+    "value": TEST_VALUE,
+    "comment": TEST_COMMENT,
+}
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_data():
+    # Clean up any leftover state from a prior aborted run before starting.
+    try:
+        properties.delete_property(TEST_NAME, TEST_OFFICE, TEST_CATEGORY)
+    except Exception:
+        pass
+
+    properties.create_property(TEST_PROPERTY_DATA)
+    yield
+    try:
+        properties.delete_property(TEST_NAME, TEST_OFFICE, TEST_CATEGORY)
+    except Exception:
+        pass
+
+
+@pytest.fixture(autouse=True)
+def init_session():
+    # Session initialization is handled by environment or global config in CDA tests
+    print("Initializing CWMS API session for properties tests...")
+
+
+def test_create_property():
+    properties.create_property(TEST_PROPERTY_DATA)
+    props = properties.get_property(TEST_NAME, TEST_OFFICE, TEST_CATEGORY)
+    assert props.json.get("name") == TEST_NAME
+    assert props.json.get("office-id") == TEST_OFFICE
+    assert props.json.get("category") == TEST_CATEGORY
+    assert props.json.get("value") == TEST_VALUE
+    assert props.json.get("comment") == TEST_COMMENT
+
+
+def test_get_properties():
+    data = properties.get_properties(
+        office_mask=TEST_OFFICE, category_id_mask=TEST_CATEGORY
+    )
+    assert data is not None
+    # Check if our test property is in the returned list
+    found = False
+    for prop in data.json:
+        if prop.get("name") == TEST_NAME and prop.get("office-id") == TEST_OFFICE:
+            found = True
+            break
+    assert found
+
+
+def test_get_property():
+    data = properties.get_property(TEST_NAME, TEST_OFFICE, TEST_CATEGORY)
+    assert data is not None
+    assert data.json.get("name") == TEST_NAME
+    assert data.json.get("office-id") == TEST_OFFICE
+    assert data.json.get("category") == TEST_CATEGORY
+    assert data.json.get("value") == TEST_VALUE
+
+
+def test_update_property():
+    updated_value = "UpdatedPytestValue"
+    updated_data = TEST_PROPERTY_DATA.copy()
+    updated_data["value"] = updated_value
+
+    properties.update_property(TEST_NAME, updated_data)
+
+    data = properties.get_property(TEST_NAME, TEST_OFFICE, TEST_CATEGORY)
+    assert data.json.get("value") == updated_value
+
+
+def test_delete_property():
+    # Create a temporary property to delete
+    temp_name = "TempDeleteProperty"
+    temp_data = TEST_PROPERTY_DATA.copy()
+    temp_data["name"] = temp_name
+
+    properties.create_property(temp_data)
+
+    # Verify it was created
+    data = properties.get_property(temp_name, TEST_OFFICE, TEST_CATEGORY)
+    assert data.json.get("name") == temp_name
+
+    # Delete it
+    properties.delete_property(temp_name, TEST_OFFICE, TEST_CATEGORY)
+
+    # Verify it was deleted
+    with pytest.raises(ApiError):
+        properties.get_property(temp_name, TEST_OFFICE, TEST_CATEGORY)

tests/resources/properties.json

@@ -0,0 +1,16 @@
+[
+    {
+        "name": "TestProperty1",
+        "office-id": "SWT",
+        "category-id": "TestCategory",
+        "value": "Value1",
+        "comment": "Comment1"
+    },
+    {
+        "name": "TestProperty2",
+        "office-id": "SWT",
+        "category-id": "TestCategory",
+        "value": "Value2",
+        "comment": "Comment2"
+    }
+]

tests/resources/property.json

@@ -0,0 +1,7 @@
+{
+    "name": "TestProperty1",
+    "office-id": "SWT",
+    "category": "TestCategory",
+    "value": "Value1",
+    "comment": "Comment1"
+}