From 1223613a7baecf105d7d2bfb35a3aac5b25a4542 Mon Sep 17 00:00:00 2001 From: Eric Novotny Date: Thu, 16 Oct 2025 11:46:40 -0500 Subject: [PATCH 1/4] levels tests --- cwms/levels/location_levels.py | 29 +++ tests/cda/levels/location_levels_test_cda.py | 195 ++++++++++++++++++ tests/cda/levels/specified_levels_test_cda.py | 93 +++++++++ tests/cda/resources/location_level.json | 11 + tests/cda/resources/specified_level.json | 5 + tests/cda/resources/specified_levels.json | 61 ++++++ .../cda/timeseries/timeseries_groups_test.py | 4 +- 7 files changed, 396 insertions(+), 2 deletions(-) create mode 100644 tests/cda/levels/location_levels_test_cda.py create mode 100644 tests/cda/levels/specified_levels_test_cda.py create mode 100644 tests/cda/resources/location_level.json create mode 100644 tests/cda/resources/specified_level.json create mode 100644 tests/cda/resources/specified_levels.json diff --git a/cwms/levels/location_levels.py b/cwms/levels/location_levels.py index d3aea2e9..4c7e6958 100644 --- a/cwms/levels/location_levels.py +++ b/cwms/levels/location_levels.py @@ -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, diff --git a/tests/cda/levels/location_levels_test_cda.py b/tests/cda/levels/location_levels_test_cda.py new file mode 100644 index 00000000..6f45c460 --- /dev/null +++ b/tests/cda/levels/location_levels_test_cda.py @@ -0,0 +1,195 @@ +# 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, timedelta +from pathlib import Path + +import pytest +import pytz + +import cwms +import cwms.levels.location_levels as location_levels +import cwms.levels.specified_levels as specified_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 = pytz.UTC.localize(datetime.utcnow()) + + +@pytest.fixture(autouse=True) +def init_session(): + print("Initializing CWMS API session for location levels tests...") + + +@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 specified level with id = "Bottom of Inlet" + specified_level_data = { + "id": "Bottom of Inlet", + "office-id": TEST_OFFICE, + "description": "Test Specified Level for Location Level Tests", + } + specified_levels.store_specified_level(specified_level_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, + effective_date=TEST_EFFECTIVE_DATE, + cascade_delete=True, + ) + # Delete the specified level + specified_levels.delete_specified_level("Bottom of Inlet", TEST_OFFICE) + # Delete the location + locations.delete_location("LEVEL", TEST_OFFICE, cascade_delete=True) + + +def test_retrieve_loc_levels_default(): + levels = location_levels.get_location_levels() + assert isinstance(levels.json, list) + + +def test_get_loc_levels(): + begin = TEST_EFFECTIVE_DATE - timedelta(days=1) + end = TEST_EFFECTIVE_DATE + timedelta(days=1) + levels = location_levels.get_location_levels( + level_id_mask=TEST_LEVEL_ID, + office_id=TEST_OFFICE, + unit=TEST_UNIT, + datum=TEST_DATUM, + begin=begin, + end=end, + page=None, + page_size=10, + ) + assert any(lvl.get("level-id") == TEST_LEVEL_ID for lvl in levels.json) + # Test DataFrame output + df = levels.df + assert not df.empty + assert TEST_LEVEL_ID in df["level-id"].values + + +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, + unit=TEST_UNIT, + ) + assert level.json.get("level-id") == TEST_LEVEL_ID + # Test DataFrame output + df = level.df + assert not df.empty + assert TEST_LEVEL_ID in df["level-id"].values + + +def test_store_loc_level_json(): + # Try storing again with a different value + data = TEST_LEVEL_DATA.copy() + data["value"] = 101.0 + location_levels.store_location_level(data) + 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("value") == 101.0 + + +def test_delete_loc_level(): + # Store a new level, then delete it + temp_level_id = "pytest_level_delete" + temp_effective_date = TEST_EFFECTIVE_DATE + timedelta(minutes=1) + data = TEST_LEVEL_DATA.copy() + data["level-id"] = temp_level_id + data["effective-date"] = temp_effective_date.isoformat() + location_levels.store_location_level(data) + location_levels.delete_location_level( + location_level_id=temp_level_id, + office_id=TEST_OFFICE, + effective_date=temp_effective_date, + cascade_delete=True, + ) + # Try to get it, should raise or return None/empty + try: + level = location_levels.get_location_level( + level_id=temp_level_id, + office_id=TEST_OFFICE, + effective_date=temp_effective_date, + unit=TEST_UNIT, + ) + assert not level.json or level.json.get("level-id") != temp_level_id + 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_api(): + # Update the value using cwms.update_location_level + new_value = 300.0 + updated_data = TEST_LEVEL_DATA.copy() + updated_data["constant-value"] = new_value + cwms.update_location_level(location_level_id=TEST_LEVEL_ID, 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 diff --git a/tests/cda/levels/specified_levels_test_cda.py b/tests/cda/levels/specified_levels_test_cda.py new file mode 100644 index 00000000..675ed581 --- /dev/null +++ b/tests/cda/levels/specified_levels_test_cda.py @@ -0,0 +1,93 @@ +# 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_levels_default(): + levels = specified_levels.get_specified_levels() + assert isinstance(levels.json, list) + + +def test_get_specified_level(): + level = specified_levels.get_specified_level(TEST_SPECIFIED_LEVEL_ID, TEST_OFFICE) + assert level.json.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("*", TEST_OFFICE) + ids = [lvl.get("specified-level-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["specified-level-id"].values + assert second_id in df["specified-level-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 + data = TEST_SPECIFIED_LEVEL_DATA.copy() + data["value"] = 456.78 + specified_levels.store_specified_level(data) + levels = specified_levels.get_specified_levels(TEST_SPECIFIED_LEVEL_ID, TEST_OFFICE) + assert any(lvl.get("value") == 456.78 for lvl in levels.json) + + +def test_delete_specified_level(): + # Store a new specified level, then delete it + temp_id = "pytest_specified_level_delete" + data = TEST_SPECIFIED_LEVEL_DATA.copy() + data["specified-level-id"] = temp_id + specified_levels.store_specified_level(data) + specified_levels.delete_specified_level(temp_id, TEST_OFFICE) + # Try to get it, should not be present + levels = specified_levels.get_specified_levels(temp_id, TEST_OFFICE) + assert not any(lvl.get("specified-level-id") == temp_id for lvl in levels.json) diff --git a/tests/cda/resources/location_level.json b/tests/cda/resources/location_level.json new file mode 100644 index 00000000..92189011 --- /dev/null +++ b/tests/cda/resources/location_level.json @@ -0,0 +1,11 @@ +{ + "location-level-id": "LEVEL.Elev.Inst.0.Bottom of Inlet", + "office-id": "SWT", + "specified-level-id": "Bottom of Inlet", + "parameter-type-id": "Inst", + "parameter-id": "Elev", + "constant-value": 145.6944, + "level-units-id": "m", + "level-date": "1900-01-01T06:00:00Z", + "duration-id": "0" +} diff --git a/tests/cda/resources/specified_level.json b/tests/cda/resources/specified_level.json new file mode 100644 index 00000000..0eb320d7 --- /dev/null +++ b/tests/cda/resources/specified_level.json @@ -0,0 +1,5 @@ +{ + "id": "Bottom of Exclusive Flood Control", + "office-id": "CWMS", + "description": "Bottom of Exclusive Flood Control Level" +} diff --git a/tests/cda/resources/specified_levels.json b/tests/cda/resources/specified_levels.json new file mode 100644 index 00000000..d65c89b3 --- /dev/null +++ b/tests/cda/resources/specified_levels.json @@ -0,0 +1,61 @@ +[ + { + "office-id": "LRL", + "id": "Guide Curve" + }, + { + "office-id": "NWDM", + "id": "Top of Summer Target" + }, + { + "office-id": "NWDM", + "id": "Top of Winter Target" + }, + { + "office-id": "MVP", + "id": "Bottom of Conservation" + }, + { + "office-id": "MVP", + "id": "Gage Zero" + }, + { + "office-id": "MVP", + "id": "Warning" + }, + { + "office-id": "SAM", + "id": "Lock Closed High Water" + }, + { + "office-id": "SAM", + "id": "Bottom of Induced Surcharge" + }, + { + "office-id": "NWDP", + "id": "USACE_Biological_Reference" + }, + { + "office-id": "NAB", + "id": "NWS Flood Stage" + }, + { + "office-id": "MVR", + "id": "TempSL" + }, + { + "office-id": "SPA", + "id": "Daily Pan Evap", + "description": "Long term Daily average" + }, + { + "office-id": "NWDP", + "id": "Injunction-Target", + "description": "Injunction Target" + }, + { + "office-id": "NWDP", + "id": "Pool-Operating", + "description": "Pool-Operating" + } +] diff --git a/tests/cda/timeseries/timeseries_groups_test.py b/tests/cda/timeseries/timeseries_groups_test.py index 109eb028..28a65366 100644 --- a/tests/cda/timeseries/timeseries_groups_test.py +++ b/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(): From a026c21c84f75d6ca6cb2d36dd71b8c5b6cb2dc2 Mon Sep 17 00:00:00 2001 From: Eric Novotny Date: Thu, 16 Oct 2025 14:03:13 -0700 Subject: [PATCH 2/4] update tests --- cwms/levels/location_levels.py | 10 +- ...est_cda.py => location_levels_cda_test.py} | 105 +++++++----------- ...st_cda.py => specified_levels_cda_test.py} | 58 ++++++---- tests/cda/resources/location_level.json | 4 +- tests/cda/resources/specified_level.json | 6 +- tests/cda/resources/specified_levels.json | 61 ---------- 6 files changed, 88 insertions(+), 156 deletions(-) rename tests/cda/levels/{location_levels_test_cda.py => location_levels_cda_test.py} (68%) rename tests/cda/levels/{specified_levels_test_cda.py => specified_levels_cda_test.py} (58%) delete mode 100644 tests/cda/resources/specified_levels.json diff --git a/cwms/levels/location_levels.py b/cwms/levels/location_levels.py index 4c7e6958..5f3f7554 100644 --- a/cwms/levels/location_levels.py +++ b/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( diff --git a/tests/cda/levels/location_levels_test_cda.py b/tests/cda/levels/location_levels_cda_test.py similarity index 68% rename from tests/cda/levels/location_levels_test_cda.py rename to tests/cda/levels/location_levels_cda_test.py index 6f45c460..3b0234fc 100644 --- a/tests/cda/levels/location_levels_test_cda.py +++ b/tests/cda/levels/location_levels_cda_test.py @@ -4,15 +4,13 @@ # Source may not be released without written approval from HEC import json -from datetime import datetime, timedelta +from datetime import timedelta from pathlib import Path +import pandas as pd import pytest -import pytz -import cwms import cwms.levels.location_levels as location_levels -import cwms.levels.specified_levels as specified_levels import cwms.locations.physical_locations as locations # Load test location level from tests/cda/resources/location_level.json @@ -24,12 +22,8 @@ TEST_OFFICE = TEST_LEVEL_DATA["office-id"] TEST_UNIT = "SI" TEST_DATUM = "NAVD88" -TEST_EFFECTIVE_DATE = pytz.UTC.localize(datetime.utcnow()) - - -@pytest.fixture(autouse=True) -def init_session(): - print("Initializing CWMS API session for location levels tests...") +TEST_EFFECTIVE_DATE = pd.to_datetime(TEST_LEVEL_DATA["level-date"]) +# datetime.now(timezone.utc) @pytest.fixture(scope="module", autouse=True) @@ -52,106 +46,89 @@ def setup_level(): } locations.store_location(BASE_LOCATION_DATA) - # Store a specified level with id = "Bottom of Inlet" - specified_level_data = { - "id": "Bottom of Inlet", - "office-id": TEST_OFFICE, - "description": "Test Specified Level for Location Level Tests", - } - specified_levels.store_specified_level(specified_level_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, - effective_date=TEST_EFFECTIVE_DATE, - cascade_delete=True, + location_level_id=TEST_LEVEL_ID, office_id=TEST_OFFICE, cascade_delete=True ) - # Delete the specified level - specified_levels.delete_specified_level("Bottom of Inlet", TEST_OFFICE) # Delete the location locations.delete_location("LEVEL", TEST_OFFICE, cascade_delete=True) -def test_retrieve_loc_levels_default(): - levels = location_levels.get_location_levels() - assert isinstance(levels.json, list) +@pytest.fixture(autouse=True) +def init_session(): + print("Initializing CWMS API session for location levels tests...") -def test_get_loc_levels(): - begin = TEST_EFFECTIVE_DATE - timedelta(days=1) - end = TEST_EFFECTIVE_DATE + timedelta(days=1) - levels = location_levels.get_location_levels( - level_id_mask=TEST_LEVEL_ID, +def test_get_loc_level(): + level = location_levels.get_location_level( + level_id=TEST_LEVEL_ID, office_id=TEST_OFFICE, - unit=TEST_UNIT, - datum=TEST_DATUM, - begin=begin, - end=end, - page=None, - page_size=10, + effective_date=TEST_EFFECTIVE_DATE, + # unit=TEST_UNIT ) - assert any(lvl.get("level-id") == TEST_LEVEL_ID for lvl in levels.json) + assert level.json.get("location-level-id") == TEST_LEVEL_ID # Test DataFrame output - df = levels.df + df = level.df assert not df.empty - assert TEST_LEVEL_ID in df["level-id"].values + assert TEST_LEVEL_ID in df["location-level-id"].values -def test_get_loc_level(): - level = location_levels.get_location_level( - level_id=TEST_LEVEL_ID, +def test_get_loc_levels(): + # begin = TEST_EFFECTIVE_DATE - timedelta(days=1) + # end = TEST_EFFECTIVE_DATE + timedelta(days=1) + levels = location_levels.get_location_levels( office_id=TEST_OFFICE, - effective_date=TEST_EFFECTIVE_DATE, - unit=TEST_UNIT, ) - assert level.json.get("level-id") == TEST_LEVEL_ID + ids = [lvl.get("location-level-id") for lvl in levels.json["levels"]] + assert TEST_LEVEL_ID in ids # Test DataFrame output - df = level.df + df = levels.df assert not df.empty - assert TEST_LEVEL_ID in df["level-id"].values + 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["value"] = 101.0 + 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=TEST_EFFECTIVE_DATE, + effective_date=new_effective_date, unit=TEST_UNIT, ) - assert level.json.get("value") == 101.0 + print(level.df) + assert level.df.loc[0, "constant-value"] == 101 def test_delete_loc_level(): # Store a new level, then delete it - temp_level_id = "pytest_level_delete" - temp_effective_date = TEST_EFFECTIVE_DATE + timedelta(minutes=1) + temp_effective_date = TEST_EFFECTIVE_DATE + timedelta(days=5) data = TEST_LEVEL_DATA.copy() - data["level-id"] = temp_level_id - data["effective-date"] = temp_effective_date.isoformat() + 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=temp_level_id, + location_level_id=TEST_LEVEL_ID, office_id=TEST_OFFICE, effective_date=temp_effective_date, - cascade_delete=True, ) # Try to get it, should raise or return None/empty try: level = location_levels.get_location_level( - level_id=temp_level_id, + level_id=TEST_LEVEL_ID, office_id=TEST_OFFICE, effective_date=temp_effective_date, unit=TEST_UNIT, ) - assert not level.json or level.json.get("level-id") != temp_level_id + assert level.df.empty except Exception: pass @@ -179,12 +156,14 @@ def test_get_loc_level_ts(): assert df["value"].notnull().any() -def test_update_location_level_api(): +def test_update_location_level(): # Update the value using cwms.update_location_level - new_value = 300.0 + new_value = 300 updated_data = TEST_LEVEL_DATA.copy() updated_data["constant-value"] = new_value - cwms.update_location_level(location_level_id=TEST_LEVEL_ID, data=updated_data) + 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, diff --git a/tests/cda/levels/specified_levels_test_cda.py b/tests/cda/levels/specified_levels_cda_test.py similarity index 58% rename from tests/cda/levels/specified_levels_test_cda.py rename to tests/cda/levels/specified_levels_cda_test.py index 675ed581..aef579aa 100644 --- a/tests/cda/levels/specified_levels_test_cda.py +++ b/tests/cda/levels/specified_levels_cda_test.py @@ -35,14 +35,11 @@ def setup_specified_level(): specified_levels.delete_specified_level(TEST_SPECIFIED_LEVEL_ID, TEST_OFFICE) -def test_get_specified_levels_default(): - levels = specified_levels.get_specified_levels() - assert isinstance(levels.json, list) - - def test_get_specified_level(): - level = specified_levels.get_specified_level(TEST_SPECIFIED_LEVEL_ID, TEST_OFFICE) - assert level.json.get("id") == TEST_SPECIFIED_LEVEL_ID + 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 @@ -58,15 +55,15 @@ def test_get_specified_levels(): specified_levels.store_specified_level(second_data) try: - levels = specified_levels.get_specified_levels("*", TEST_OFFICE) - ids = [lvl.get("specified-level-id") for lvl in levels.json] + 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["specified-level-id"].values - assert second_id in df["specified-level-id"].values + 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) @@ -74,20 +71,37 @@ def test_get_specified_levels(): 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["value"] = 456.78 - specified_levels.store_specified_level(data) - levels = specified_levels.get_specified_levels(TEST_SPECIFIED_LEVEL_ID, TEST_OFFICE) - assert any(lvl.get("value") == 456.78 for lvl in levels.json) + 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_specified_level_delete" + temp_id = "pytest delete" data = TEST_SPECIFIED_LEVEL_DATA.copy() - data["specified-level-id"] = temp_id - specified_levels.store_specified_level(data) - specified_levels.delete_specified_level(temp_id, TEST_OFFICE) - # Try to get it, should not be present - levels = specified_levels.get_specified_levels(temp_id, TEST_OFFICE) - assert not any(lvl.get("specified-level-id") == temp_id for lvl in levels.json) + 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 diff --git a/tests/cda/resources/location_level.json b/tests/cda/resources/location_level.json index 92189011..d53e3593 100644 --- a/tests/cda/resources/location_level.json +++ b/tests/cda/resources/location_level.json @@ -1,11 +1,11 @@ { "location-level-id": "LEVEL.Elev.Inst.0.Bottom of Inlet", - "office-id": "SWT", + "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": "1900-01-01T06:00:00Z", + "level-date": "2000-01-01T06:00:00Z", "duration-id": "0" } diff --git a/tests/cda/resources/specified_level.json b/tests/cda/resources/specified_level.json index 0eb320d7..e08b36cc 100644 --- a/tests/cda/resources/specified_level.json +++ b/tests/cda/resources/specified_level.json @@ -1,5 +1,5 @@ { - "id": "Bottom of Exclusive Flood Control", - "office-id": "CWMS", - "description": "Bottom of Exclusive Flood Control Level" + "id": "MVP Bottom of Exclusive Flood Control", + "office-id": "MVP", + "description": "MVP Bottom of Exclusive Flood Control Level" } diff --git a/tests/cda/resources/specified_levels.json b/tests/cda/resources/specified_levels.json deleted file mode 100644 index d65c89b3..00000000 --- a/tests/cda/resources/specified_levels.json +++ /dev/null @@ -1,61 +0,0 @@ -[ - { - "office-id": "LRL", - "id": "Guide Curve" - }, - { - "office-id": "NWDM", - "id": "Top of Summer Target" - }, - { - "office-id": "NWDM", - "id": "Top of Winter Target" - }, - { - "office-id": "MVP", - "id": "Bottom of Conservation" - }, - { - "office-id": "MVP", - "id": "Gage Zero" - }, - { - "office-id": "MVP", - "id": "Warning" - }, - { - "office-id": "SAM", - "id": "Lock Closed High Water" - }, - { - "office-id": "SAM", - "id": "Bottom of Induced Surcharge" - }, - { - "office-id": "NWDP", - "id": "USACE_Biological_Reference" - }, - { - "office-id": "NAB", - "id": "NWS Flood Stage" - }, - { - "office-id": "MVR", - "id": "TempSL" - }, - { - "office-id": "SPA", - "id": "Daily Pan Evap", - "description": "Long term Daily average" - }, - { - "office-id": "NWDP", - "id": "Injunction-Target", - "description": "Injunction Target" - }, - { - "office-id": "NWDP", - "id": "Pool-Operating", - "description": "Pool-Operating" - } -] From 1e534c21db87de05d6320779ae3c892a312e7494 Mon Sep 17 00:00:00 2001 From: Eric Novotny Date: Thu, 16 Oct 2025 14:08:17 -0700 Subject: [PATCH 3/4] remove mock tests --- tests/cda/levels/location_levels_cda_test.py | 3 - tests/mock/levels/location_levels_test.py | 113 ------------------- tests/mock/levels/specified_levels_test.py | 72 ------------ 3 files changed, 188 deletions(-) delete mode 100644 tests/mock/levels/location_levels_test.py delete mode 100644 tests/mock/levels/specified_levels_test.py diff --git a/tests/cda/levels/location_levels_cda_test.py b/tests/cda/levels/location_levels_cda_test.py index 3b0234fc..f74aa9d2 100644 --- a/tests/cda/levels/location_levels_cda_test.py +++ b/tests/cda/levels/location_levels_cda_test.py @@ -23,7 +23,6 @@ TEST_UNIT = "SI" TEST_DATUM = "NAVD88" TEST_EFFECTIVE_DATE = pd.to_datetime(TEST_LEVEL_DATA["level-date"]) -# datetime.now(timezone.utc) @pytest.fixture(scope="module", autouse=True) @@ -77,8 +76,6 @@ def test_get_loc_level(): def test_get_loc_levels(): - # begin = TEST_EFFECTIVE_DATE - timedelta(days=1) - # end = TEST_EFFECTIVE_DATE + timedelta(days=1) levels = location_levels.get_location_levels( office_id=TEST_OFFICE, ) diff --git a/tests/mock/levels/location_levels_test.py b/tests/mock/levels/location_levels_test.py deleted file mode 100644 index f73afbc4..00000000 --- a/tests/mock/levels/location_levels_test.py +++ /dev/null @@ -1,113 +0,0 @@ -# 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 - -from datetime import datetime - -import pytest -import pytz - -import cwms.api -import cwms.levels.location_levels as location_levels -from tests._test_utils import read_resource_file - -_MOCK_ROOT = "https://mockwebserver.cwms.gov" -_LOC_LEVELS_JSON = read_resource_file("location_levels.json") -_LOC_LEVEL_JSON = read_resource_file("location_level.json") -_LOC_LEVEL_TS_JSON = read_resource_file("level_timeseries.json") - - -@pytest.fixture(autouse=True) -def init_session(): - cwms.api.init_session(api_root=_MOCK_ROOT) - - -def test_retrieve_loc_levels_default(requests_mock): - requests_mock.get( - f"{_MOCK_ROOT}/levels?level-id-mask=%2A", - json=_LOC_LEVELS_JSON, - ) - levels = location_levels.get_location_levels() - assert levels.json == _LOC_LEVELS_JSON - - -def test_get_loc_levels(requests_mock): - requests_mock.get( - f"{_MOCK_ROOT}/levels?office=SWT&level-id-mask=AARK.Elev.Inst.0.Bottom+of+Inlet&" - "unit=m&datum=NAV88&begin=2020-02-14T10%3A30%3A00-08%3A00&end=2020-03-30T10%3A30%3A00-07%3A00&" - "page=MHx8bnVsbHx8MTAw&page-size=100", - json=_LOC_LEVELS_JSON, - ) - - level_id = "AARK.Elev.Inst.0.Bottom of Inlet" - office_id = "SWT" - unit = "m" - datum = "NAV88" - timezone = pytz.timezone("US/Pacific") - begin = timezone.localize(datetime(2020, 2, 14, 10, 30, 0)) - end = timezone.localize(datetime(2020, 3, 30, 10, 30, 0)) - page = "MHx8bnVsbHx8MTAw" - page_size = 100 - levels = location_levels.get_location_levels( - level_id, office_id, unit, datum, begin, end, page, page_size - ) - assert levels.json == _LOC_LEVELS_JSON - - -def test_get_loc_level(requests_mock): - requests_mock.get( - f"{_MOCK_ROOT}/levels/AARK.Elev.Inst.0.Bottom%20of%20Inlet?office=SWT&" - "unit=m&effective-date=2020-02-14T10%3A30%3A00-08%3A00", - json=_LOC_LEVEL_JSON, - ) - level_id = "AARK.Elev.Inst.0.Bottom of Inlet" - office_id = "SWT" - unit = "m" - timezone = pytz.timezone("US/Pacific") - effective_date = timezone.localize(datetime(2020, 2, 14, 10, 30, 0)) - levels = location_levels.get_location_level( - level_id, office_id, effective_date, unit - ) - assert levels.json == _LOC_LEVEL_JSON - - -def test_store_loc_level_json(requests_mock): - requests_mock.post(f"{_MOCK_ROOT}/levels") - data = _LOC_LEVEL_JSON - location_levels.store_location_level(data) - assert requests_mock.called - assert requests_mock.call_count == 1 - - -def test_delete_loc_level(requests_mock): - requests_mock.delete( - f"{_MOCK_ROOT}/levels/AARK.Elev.Inst.0.Bottom%20of%20Inlet?office=SWT&" - "effective-date=2020-02-14T10%3A30%3A00-08%3A00&cascade-delete=True", - json=_LOC_LEVEL_JSON, - ) - level_id = "AARK.Elev.Inst.0.Bottom of Inlet" - office_id = "SWT" - timezone = pytz.timezone("US/Pacific") - effective_date = timezone.localize(datetime(2020, 2, 14, 10, 30, 0)) - location_levels.delete_location_level(level_id, office_id, effective_date, True) - assert requests_mock.called - assert requests_mock.call_count == 1 - - -def test_get_loc_level_ts(requests_mock): - requests_mock.get( - f"{_MOCK_ROOT}/levels/AARK.Elev.Inst.0.Bottom%20of%20Inlet/timeseries?office=SWT&unit=m&" - "begin=2020-02-14T10%3A30%3A00-08%3A00&end=2020-03-14T10%3A30%3A00-07%3A00&interval=1Day", - json=_LOC_LEVEL_TS_JSON, - ) - level_id = "AARK.Elev.Inst.0.Bottom of Inlet" - office_id = "SWT" - interval = "1Day" - timezone = pytz.timezone("US/Pacific") - begin = timezone.localize(datetime(2020, 2, 14, 10, 30, 0)) - end = timezone.localize(datetime(2020, 3, 14, 10, 30, 0)) - levels = location_levels.get_level_as_timeseries( - level_id, office_id, "m", begin, end, interval - ) - assert levels.json == _LOC_LEVEL_TS_JSON diff --git a/tests/mock/levels/specified_levels_test.py b/tests/mock/levels/specified_levels_test.py deleted file mode 100644 index 19d32a36..00000000 --- a/tests/mock/levels/specified_levels_test.py +++ /dev/null @@ -1,72 +0,0 @@ -# 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 - -from datetime import datetime - -import pytest -import pytz -from requests.exceptions import HTTPError - -import cwms.api -import cwms.levels.specified_levels as specified_levels -from tests._test_utils import read_resource_file - -_MOCK_ROOT = "https://mockwebserver.cwms.gov" -_SPEC_LEVELS_JSON = read_resource_file("specified_levels.json") -_SPEC_LEVEL_JSON = read_resource_file("specified_level.json") - - -@pytest.fixture(autouse=True) -def init_session(): - cwms.api.init_session(api_root=_MOCK_ROOT) - - -def test_get_specified_levels_default(requests_mock): - requests_mock.get( - f"{_MOCK_ROOT}/specified-levels?office=%2A&template-id-mask=%2A", - json=_SPEC_LEVELS_JSON, - ) - levels = specified_levels.get_specified_levels() - assert levels.json == _SPEC_LEVELS_JSON - - -def test_get_specified_levels(requests_mock): - requests_mock.get( - f"{_MOCK_ROOT}/specified-levels?office=SWT&template-id-mask=%2A", - json=_SPEC_LEVELS_JSON, - ) - levels = specified_levels.get_specified_levels("*", "SWT") - assert levels.json == _SPEC_LEVELS_JSON - - -def test_store_specified_level(requests_mock): - requests_mock.post( - f"{_MOCK_ROOT}/specified-levels?fail-if-exists=True", - status_code=200, - json=_SPEC_LEVEL_JSON, - ) - specified_levels.store_specified_level(_SPEC_LEVEL_JSON) - assert requests_mock.called - assert requests_mock.call_count == 1 - - -def test_delete_specified_level(requests_mock): - requests_mock.delete( - f"{_MOCK_ROOT}/specified-levels/Test?office=SWT", - status_code=200, - ) - specified_levels.delete_specified_level("Test", "SWT") - assert requests_mock.called - assert requests_mock.call_count == 1 - - -def test_update_specified_level(requests_mock): - requests_mock.patch( - f"{_MOCK_ROOT}/specified-levels/Test?specified-level-id=TEst2&office=SWT", - status_code=200, - ) - specified_levels.update_specified_level("Test", "Test2", "SWT") - assert requests_mock.called - assert requests_mock.call_count == 1 From dc0b3a9d62089dbf73a7fd8b07656b7ddb12f7e0 Mon Sep 17 00:00:00 2001 From: Eric Novotny Date: Thu, 16 Oct 2025 14:15:23 -0700 Subject: [PATCH 4/4] remove comment --- tests/cda/levels/location_levels_cda_test.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/cda/levels/location_levels_cda_test.py b/tests/cda/levels/location_levels_cda_test.py index f74aa9d2..6ebacc7a 100644 --- a/tests/cda/levels/location_levels_cda_test.py +++ b/tests/cda/levels/location_levels_cda_test.py @@ -66,7 +66,6 @@ def test_get_loc_level(): level_id=TEST_LEVEL_ID, office_id=TEST_OFFICE, effective_date=TEST_EFFECTIVE_DATE, - # unit=TEST_UNIT ) assert level.json.get("location-level-id") == TEST_LEVEL_ID # Test DataFrame output