diff --git a/cwms/levels/location_levels.py b/cwms/levels/location_levels.py index d3aea2e9..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( @@ -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_cda_test.py b/tests/cda/levels/location_levels_cda_test.py new file mode 100644 index 00000000..6ebacc7a --- /dev/null +++ b/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 diff --git a/tests/cda/levels/specified_levels_cda_test.py b/tests/cda/levels/specified_levels_cda_test.py new file mode 100644 index 00000000..aef579aa --- /dev/null +++ b/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 diff --git a/tests/cda/resources/location_level.json b/tests/cda/resources/location_level.json new file mode 100644 index 00000000..d53e3593 --- /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": "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" +} diff --git a/tests/cda/resources/specified_level.json b/tests/cda/resources/specified_level.json new file mode 100644 index 00000000..e08b36cc --- /dev/null +++ b/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" +} 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(): 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