Commit awqms test suite

pyproject.toml

@@ -5,12 +5,10 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
-    "aiohttp>=3.11.11",
     "dagster-webserver>=1.9.4",
     "dagster>=1.9.4",
     "httpx>=0.28.1",
     "requests>=2.32.3",
-    "pytest>=8.3.4",
     "pandas>=2.2.3",
     "frost-sta-client>=1.1.45",
     "debugpy>=1.8.11",
@@ -25,6 +23,14 @@ dependencies = [
 [tool.setuptools]
 packages = ["userCode"]
 
+[tool.setuptools.extra_requirements]
+dev = [
+    "pytest>=8.3.4",
+    "pytest-asyncio==0.25.2",
+    "pytest-mock==3.14.0",
+    "requests-mock==1.12.1"
+]
+
 [tool.dagster]
 module_name = "userCode"
 code_location_name = "userCode"

tests/awqms/__init__.py

@@ -0,0 +1,7 @@
+# =================================================================
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================

tests/awqms/conftest.py

@@ -0,0 +1,99 @@
+# =================================================================
+#
+# Authors: Ben Webb <[email protected]>
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+from datetime import datetime
+import pytest
+
+from userCode.awqms.types import StationData, GmlPoint, ResultSummary
+from userCode.types import Datastream, Observation
+
+
+@pytest.fixture
+def sample_station_data():
+    return StationData(
+        MonitoringLocationId="12005-ORDEQ",
+        MonitoringLocationName="McKay Creek at Kirk Road (Pendleton)",
+        MonitoringLocationType="River/Stream",
+        OrganizationIdentifier="OREGONDEQ",
+        WaterbodyName="McKay Creekr",
+        CountyName="Umatilla",
+        Huc8="17070103",
+        Huc12="170701030408",
+        Geometry=GmlPoint(longitude=-118.8239942, latitude=45.65429575),
+        Datastreams=[
+            ResultSummary(
+                activity_type="Field Msr/Obs",
+                observed_property="Temperature, water"
+            )
+        ]
+    )
+
+
+@pytest.fixture
+def sample_datastream():
+    return Datastream(
+        **{
+            "@iot.id": "12005-ORDEQ-2714",
+            "name": "Temperature, water",
+            "description": "Temperature, water",
+            "observationType": "http://www.opengis.net/def/observationType/OGC-OM/2.0/OM_Measurement",
+            "unitOfMeasurement": {
+                "name": "celsius",
+                "symbol": "°C",
+                "definition": "degree celsius"
+            },
+            "ObservedProperty": {
+                "@iot.id": 12,
+                "name": "Temperature, water",
+                "description": "Temperature of water in celsius",
+                "definition": "http://vocabulary.odm2.org/variablename/temperature/",
+                "properties": {
+                    "uri": "http://vocabulary.odm2.org/variablename/temperature/"
+                }
+            },
+            "Sensor": {
+                "@iot.id": 0,
+                "name": "Unknown",
+                "description": "Unknown",
+                "encodingType": "Unknown",
+                "metadata": "Unknown"
+            },
+            "Thing": {
+                "@iot.id": "12005-ORDEQ"
+            }
+        }
+    )
+
+
+@pytest.fixture
+def sample_observation(sample_datastream, sample_station_data):
+    return Observation(
+        **{
+            "@iot.id": 1234,
+            "result": 20.5,
+            "phenomenonTime": datetime.now().isoformat(),
+            "resultTime": datetime.now().isoformat(),
+            "Datastream": {
+                "@iot.id": sample_datastream.iotid
+            },
+            "FeatureOfInterest": {
+                "name": sample_station_data.MonitoringLocationName,
+                "description": "Monitoring Location",
+                "encodingType": "application/vnd.geo+json",
+                "feature": {
+                    "type": "Point",
+                    "coordinates": [
+                        sample_station_data.Geometry.longitude,
+                        sample_station_data.Geometry.latitude
+                    ]
+                }
+            },
+        }
+    )

tests/awqms/test_e2e.py

@@ -0,0 +1,95 @@
+# =================================================================
+#
+# Authors: Ben Webb <[email protected]>
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+import pytest
+from unittest.mock import patch
+
+from userCode.awqms.dag import (
+    awqms_preflight_checks, post_awqms_station,
+    post_awqms_datastreams, awqms_datastreams,
+    batch_post_awqms_observations, awqms_observations,
+    awqms_schedule
+)
+from userCode.helper_classes import BatchHelper
+
+
+def test_awqms_preflight_checks():
+    with patch('requests.get') as mock_get:
+        mock_get.return_value.ok = True
+        result = awqms_preflight_checks()
+        assert result is None
+
+    with patch('requests.get') as mock_get:
+        mock_get.return_value.ok = False
+        with pytest.raises(AssertionError):
+            awqms_preflight_checks()
+
+
+def test_post_awqms_station(sample_station_data):
+    with patch('requests.get') as mock_get:
+        # Simulate station not found
+        mock_get.return_value.status_code = 500
+        with patch('requests.post') as mock_post:
+            mock_post.return_value.ok = True
+            post_awqms_station(sample_station_data)
+            mock_post.assert_called_once()
+
+
+def test_awqms_datastreams(sample_station_data):
+    datastreams = awqms_datastreams(sample_station_data)
+    assert len(datastreams) > 0  # type: ignore
+
+
+def test_post_awqms_datastreams(sample_datastream):
+    with patch('requests.get') as mock_get:
+        mock_get.return_value.status_code = 404
+        with patch('requests.post') as mock_post:
+            mock_post.return_value.ok = True
+            post_awqms_datastreams([sample_datastream])
+            mock_post.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_awqms_observations(sample_station_data, sample_datastream):
+    datastreams = [sample_datastream,]
+
+    with patch('userCode.awqms.lib.fetch_observation_ids'
+               ) as mock_fetch_observation_ids:
+        mock_fetch_observation_ids.return_value = set()
+
+        with patch('userCode.awqms.lib.fetch_observations'
+                   ) as mock_fetch_observations:
+            mock_fetch_observations.return_value = [
+                {"ResultValue": 10,
+                 "StartDateTime": "2025-01-01",
+                 "Status": "Final"}]
+
+            observations = await awqms_observations(
+                sample_station_data, datastreams)  # type: ignore
+            assert len(observations) > 0
+
+
+def test_batch_post_awqms_observations(sample_observation):
+    with patch.object(BatchHelper,
+                      'send_observations') as mock_send_observations:
+        batch_post_awqms_observations([sample_observation])
+        mock_send_observations.assert_called_once()
+
+
+def test_awqms_schedule_triggering():
+    pch = 'userCode.awqms.dag.station_partition.get_partition_keys'
+    with patch(pch) as mock_get_partition_keys:
+        mock_get_partition_keys.return_value = ["1234", "5678"]
+        schedule = awqms_schedule()
+        # Verify that RunRequest is yielded for each partition
+        runs = list(schedule)  # type: ignore
+        assert len(runs) == 2
+        assert runs[0].partition_key == "1234"
+        assert runs[1].partition_key == "5678"

tests/awqms/test_lib.py

@@ -0,0 +1,93 @@
+# =================================================================
+#
+# Authors: Ben Webb <[email protected]>
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+from pathlib import Path
+import pytest
+import tempfile
+from unittest.mock import patch, Mock
+
+from userCode.awqms.lib import read_csv, fetch_station, fetch_observations
+
+
+@pytest.fixture
+def sample_csv_path(tmp_path):
+    csv_content = "station_id\nTEST123\nTEST456\nTEST789"
+    csv_file = tmp_path / "test_stations.csv"
+    csv_file.write_text(csv_content)
+    return csv_file
+
+
+def test_read_csv_success(sample_csv_path):
+    result = read_csv(sample_csv_path)
+    assert result == ["TEST123", "TEST456", "TEST789"]
+
+
+def test_read_csv_file_not_found():
+    result = read_csv(Path("nonexistent.csv"))
+    assert result == []
+
+
+@patch('userCode.cache.ShelveCache', autospec=True)
+def test_fetch_station(mock_shelve_cache_cls):
+    with tempfile.NamedTemporaryFile() as temp_db:
+        mock_shelve_cache_cls.db = temp_db.name + ".db"
+
+        result = fetch_station("12005-ORDEQ")
+        assert len(result) == 99994
+
+
+@patch('userCode.cache.ShelveCache', autospec=True)
+def test_fetch_station_error(mock_shelve_cache_cls):
+    with tempfile.NamedTemporaryFile() as temp_db:
+        # Configure the mocked ShelveCache class to use a temporary database
+        mock_shelve_cache_cls.db = temp_db.name + ".db"
+
+        # Mock the behavior of the cache instance
+        mock_cache_instance = Mock()
+        mock_cache_instance.get_or_fetch.return_value = (b'error', 404)
+        mock_shelve_cache_cls.return_value = mock_cache_instance
+
+        # Test that a RuntimeError is raised for a failed fetch
+        with pytest.raises(RuntimeError,
+                           match="Request.*failed with status 404"):
+            fetch_station("120016-ORDEQ")
+
+
+@patch('userCode.cache.ShelveCache', autospec=True)
+def test_fetch_observations(mock_shelve_cache_cls):
+    with tempfile.NamedTemporaryFile() as temp_db:
+        # Configure the mocked ShelveCache class to use a temporary database
+        mock_shelve_cache_cls.db = temp_db.name + ".db"
+
+        # Mock a valid response with JSON data
+        mock_cache_instance = Mock()
+        mock_shelve_cache_cls.return_value = mock_cache_instance
+
+        # Fetch observations and assert the results are as expected
+        result = fetch_observations("Temperature, water", "12005-ORDEQ")
+
+        assert len(result) == 3354
+
+
+@patch('userCode.cache.ShelveCache', autospec=True)
+def test_fetch_observations_invalid_json(mock_shelve_cache_cls):
+    with tempfile.NamedTemporaryFile() as temp_db:
+        # Configure the mocked ShelveCache class to use a temporary database
+        mock_shelve_cache_cls.db = temp_db.name + ".db"
+
+        # Mock an invalid JSON response
+        mock_cache_instance = Mock()
+        mock_cache_instance.get_or_fetch.return_value = (b'invalid json', 200)
+        mock_shelve_cache_cls.return_value = mock_cache_instance
+
+        # Test that a RuntimeError is raised for invalid JSON data
+        with pytest.raises(RuntimeError,
+                           match="Request to.*failed with status 404"):
+            fetch_observations("Temperature", "TEST123")