levels tests

cwms/levels/location_levels.py

@@ -13,7 +13,7 @@
 
 
 def get_location_levels(
-    level_id_mask: str = "*",
+    level_id_mask: Optional[str] = None,
     office_id: Optional[str] = None,
     unit: Optional[str] = None,
     datum: Optional[str] = None,
@@ -58,13 +58,13 @@ def get_location_levels(
         "level-id-mask": level_id_mask,
         "unit": unit,
         "datum": datum,
-        "begin": begin.isoformat() if begin else "",
-        "end": end.isoformat() if end else "",
+        "begin": begin.isoformat() if begin else None,
+        "end": end.isoformat() if end else None,
         "page": page,
         "page-size": page_size,
     }
-    response = api.get(endpoint, params)
-    return Data(response)
+    response = api.get(endpoint=endpoint, params=params)
+    return Data(json=response, selector="levels")
 
 
 def get_location_level(
@@ -169,6 +169,35 @@ def delete_location_level(
     return api.delete(endpoint, params)
 
 
+def update_location_level(
+    data: JSON, level_id: str, effective_date: Optional[datetime] = None
+) -> None:
+    """
+    Parameters
+    ----------
+    data : dict
+        The JSON data dictionary containing the updated location level information.
+    level_id : str
+        The ID of the location level to be updated.
+    effective_date : datetime, optional
+        The effective date of the location level to be updated.
+        If the datetime has a timezone it will be used, otherwise it is assumed to be in UTC.
+
+    """
+    if data is None:
+        raise ValueError(
+            "Cannot update a location level without a JSON data dictionary"
+        )
+    if level_id is None:
+        raise ValueError("Cannot update a location level without an id")
+    endpoint = f"levels/{level_id}"
+
+    params = {
+        "effective-date": (effective_date.isoformat() if effective_date else None),
+    }
+    return api.patch(endpoint, data, params)
+
+
 def get_level_as_timeseries(
     location_level_id: str,
     office_id: str,

tests/cda/levels/location_levels_cda_test.py

@@ -0,0 +1,170 @@
+#  Copyright (c) 2024
+#  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 json
+from datetime import timedelta
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+import cwms.levels.location_levels as location_levels
+import cwms.locations.physical_locations as locations
+
+# Load test location level from tests/cda/resources/location_level.json
+LEVEL_RESOURCE_PATH = Path(__file__).parent.parent / "resources" / "location_level.json"
+with open(LEVEL_RESOURCE_PATH, "r") as f:
+    TEST_LEVEL_DATA = json.load(f)
+
+TEST_LEVEL_ID = TEST_LEVEL_DATA["location-level-id"]
+TEST_OFFICE = TEST_LEVEL_DATA["office-id"]
+TEST_UNIT = "SI"
+TEST_DATUM = "NAVD88"
+TEST_EFFECTIVE_DATE = pd.to_datetime(TEST_LEVEL_DATA["level-date"])
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_level():
+    # Store a test location with name "LEVEL"
+    BASE_LOCATION_DATA = {
+        "name": "LEVEL",
+        "office-id": TEST_OFFICE,
+        "latitude": 44.0,
+        "longitude": -93.0,
+        "elevation": 250.0,
+        "horizontal-datum": "NAD83",
+        "vertical-datum": "NAVD88",
+        "location-type": "TESTING",
+        "public-name": "Test Location",
+        "long-name": "A pytest-generated location",
+        "timezone-name": "America/Los_Angeles",
+        "location-kind": "SITE",
+        "nation": "US",
+    }
+    locations.store_location(BASE_LOCATION_DATA)
+
+    # Store a test location level before tests
+    location_levels.store_location_level(TEST_LEVEL_DATA)
+    yield
+    # Delete the test location level after tests
+    location_levels.delete_location_level(
+        location_level_id=TEST_LEVEL_ID, office_id=TEST_OFFICE, cascade_delete=True
+    )
+    # Delete the location
+    locations.delete_location("LEVEL", TEST_OFFICE, cascade_delete=True)
+
+
+@pytest.fixture(autouse=True)
+def init_session():
+    print("Initializing CWMS API session for location levels tests...")
+
+
+def test_get_loc_level():
+    level = location_levels.get_location_level(
+        level_id=TEST_LEVEL_ID,
+        office_id=TEST_OFFICE,
+        effective_date=TEST_EFFECTIVE_DATE,
+    )
+    assert level.json.get("location-level-id") == TEST_LEVEL_ID
+    # Test DataFrame output
+    df = level.df
+    assert not df.empty
+    assert TEST_LEVEL_ID in df["location-level-id"].values
+
+
+def test_get_loc_levels():
+    levels = location_levels.get_location_levels(
+        office_id=TEST_OFFICE,
+    )
+    ids = [lvl.get("location-level-id") for lvl in levels.json["levels"]]
+    assert TEST_LEVEL_ID in ids
+    # Test DataFrame output
+    df = levels.df
+    assert not df.empty
+    assert TEST_LEVEL_ID in df["location-level-id"].values
+
+
+def test_store_loc_level_json():
+    # Try storing again with a different value
+    level_date = "2010-01-01T06:00:00Z"
+    new_effective_date = pd.to_datetime("2010-01-01T06:00:00Z")
+    data = TEST_LEVEL_DATA.copy()
+    data["level-date"] = level_date
+    data["constant-value"] = 101
+    location_levels.store_location_level(data)
+    level = location_levels.get_location_level(
+        level_id=TEST_LEVEL_ID,
+        office_id=TEST_OFFICE,
+        effective_date=new_effective_date,
+        unit=TEST_UNIT,
+    )
+    print(level.df)
+    assert level.df.loc[0, "constant-value"] == 101
+
+
+def test_delete_loc_level():
+    # Store a new level, then delete it
+    temp_effective_date = TEST_EFFECTIVE_DATE + timedelta(days=5)
+    data = TEST_LEVEL_DATA.copy()
+    data["level-date"] = temp_effective_date.isoformat()
+    data["constant-value"] = 300
+    location_levels.store_location_level(data)
+    location_levels.delete_location_level(
+        location_level_id=TEST_LEVEL_ID,
+        office_id=TEST_OFFICE,
+        effective_date=temp_effective_date,
+    )
+    # Try to get it, should raise or return None/empty
+    try:
+        level = location_levels.get_location_level(
+            level_id=TEST_LEVEL_ID,
+            office_id=TEST_OFFICE,
+            effective_date=temp_effective_date,
+            unit=TEST_UNIT,
+        )
+        assert level.df.empty
+    except Exception:
+        pass
+
+
+def test_get_loc_level_ts():
+    interval = "1Day"
+    begin = TEST_EFFECTIVE_DATE - timedelta(days=2)
+    end = TEST_EFFECTIVE_DATE + timedelta(days=2)
+    ts = location_levels.get_level_as_timeseries(
+        location_level_id=TEST_LEVEL_ID,
+        office_id=TEST_OFFICE,
+        unit=TEST_UNIT,
+        begin=begin,
+        end=end,
+        interval=interval,
+    )
+    assert isinstance(ts.json, list) or ts.json is not None
+    # Test DataFrame output
+    df = ts.df
+    assert df is not None
+    assert not df.empty
+    # Check that the DataFrame contains expected columns and values
+    assert "value" in df.columns
+    # Optionally check for at least one value (if expected)
+    assert df["value"].notnull().any()
+
+
+def test_update_location_level():
+    # Update the value using cwms.update_location_level
+    new_value = 300
+    updated_data = TEST_LEVEL_DATA.copy()
+    updated_data["constant-value"] = new_value
+    location_levels.update_location_level(
+        level_id=TEST_LEVEL_ID, effective_date=TEST_EFFECTIVE_DATE, data=updated_data
+    )
+    # Retrieve and check the updated value
+    level = location_levels.get_location_level(
+        level_id=TEST_LEVEL_ID,
+        office_id=TEST_OFFICE,
+        effective_date=TEST_EFFECTIVE_DATE,
+        unit=TEST_UNIT,
+    )
+    assert level.json.get("constant-value") == new_value

tests/cda/levels/specified_levels_cda_test.py

@@ -0,0 +1,107 @@
+#  Copyright (c) 2024
+#  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 json
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+import cwms
+import cwms.levels.specified_levels as specified_levels
+
+# Load test specified level from tests/cda/resources/specified_level.json
+RESOURCE_PATH = Path(__file__).parent.parent / "resources" / "specified_level.json"
+with open(RESOURCE_PATH, "r") as f:
+    TEST_SPECIFIED_LEVEL_DATA = json.load(f)
+
+TEST_SPECIFIED_LEVEL_ID = TEST_SPECIFIED_LEVEL_DATA["id"]
+TEST_OFFICE = TEST_SPECIFIED_LEVEL_DATA["office-id"]
+
+
+@pytest.fixture(autouse=True)
+def init_session():
+    print("Initializing CWMS API session for specified levels tests...")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def setup_specified_level():
+    # Store a test specified level before tests
+    specified_levels.store_specified_level(TEST_SPECIFIED_LEVEL_DATA)
+    yield
+    # Delete the test specified level after tests
+    specified_levels.delete_specified_level(TEST_SPECIFIED_LEVEL_ID, TEST_OFFICE)
+
+
+def test_get_specified_level():
+    level = specified_levels.get_specified_levels(
+        specified_level_mask=TEST_SPECIFIED_LEVEL_ID, office_id=TEST_OFFICE
+    )
+    assert level.json[0].get("id") == TEST_SPECIFIED_LEVEL_ID
+    # DataFrame check
+    df = level.df
+    assert not df.empty
+    assert TEST_SPECIFIED_LEVEL_ID in df["id"].values
+
+
+def test_get_specified_levels():
+    # Store a second specified level
+    second_id = "pytest_specified_level_2"
+    second_data = TEST_SPECIFIED_LEVEL_DATA.copy()
+    second_data["id"] = second_id
+    second_data["office-id"] = TEST_OFFICE
+    specified_levels.store_specified_level(second_data)
+
+    try:
+        levels = specified_levels.get_specified_levels(office_id=TEST_OFFICE)
+        ids = [lvl.get("id") for lvl in levels.json]
+        assert TEST_SPECIFIED_LEVEL_ID in ids
+        assert second_id in ids
+        # DataFrame check
+        df = levels.df
+        assert not df.empty
+        assert TEST_SPECIFIED_LEVEL_ID in df["id"].values
+        assert second_id in df["id"].values
+    finally:
+        # Cleanup second specified level
+        specified_levels.delete_specified_level(second_id, TEST_OFFICE)
+
+
+def test_store_specified_level():
+    # Try storing again with a different value
+    new_specified_level_id = "MVP Test Specified"
+    new_specified_level_disc = "MVP Level"
+    data = TEST_SPECIFIED_LEVEL_DATA.copy()
+    data["id"] = new_specified_level_id
+    data["description"] = "MVP Level"
+    specified_levels.store_specified_level(data=data)
+    try:
+        levels = specified_levels.get_specified_levels(
+            specified_level_mask=new_specified_level_id, office_id=TEST_OFFICE
+        )
+        assert levels.json[0].get("description") == new_specified_level_disc
+    finally:
+        specified_levels.delete_specified_level(
+            specified_level_id=new_specified_level_id, office_id=TEST_OFFICE
+        )
+
+
+def test_delete_specified_level():
+    # Store a new specified level, then delete it
+    temp_id = "pytest delete"
+    data = TEST_SPECIFIED_LEVEL_DATA.copy()
+    data["id"] = temp_id
+    specified_levels.store_specified_level(data=data)
+    specified_levels.delete_specified_level(
+        specified_level_id=temp_id, office_id=TEST_OFFICE
+    )
+    # Try to get it, should raise or return None/empty
+    try:
+        levels = specified_levels.get_specified_levels(
+            specified_level_mask=temp_id, office_id=TEST_OFFICE
+        )
+        assert not any(lvl.get("id") == temp_id for lvl in levels.json)
+    except Exception:
+        pass

tests/cda/resources/location_level.json

@@ -0,0 +1,11 @@
+{
+  "location-level-id": "LEVEL.Elev.Inst.0.Bottom of Inlet",
+  "office-id": "MVP",
+  "specified-level-id": "Bottom of Inlet",
+  "parameter-type-id": "Inst",
+  "parameter-id": "Elev",
+  "constant-value": 145.6944,
+  "level-units-id": "m",
+  "level-date": "2000-01-01T06:00:00Z",
+  "duration-id": "0"
+}

tests/cda/resources/specified_level.json

@@ -0,0 +1,5 @@
+{
+  "id": "MVP Bottom of Exclusive Flood Control",
+  "office-id": "MVP",
+  "description": "MVP Bottom of Exclusive Flood Control Level"
+}

tests/cda/timeseries/timeseries_groups_test.py

@@ -71,8 +71,8 @@ def setup_data():
 
 
 @pytest.fixture(autouse=True)
-def init_session(request):
-    print("Initializing CWMS API session for locations operations test...")
+def init_session():
+    print("Initializing CWMS API session for timeseries groups tests...")
 
 
 def test_store_timeseries_group():