From edf4107408d42e9cb50c831c59d73604860c7d93 Mon Sep 17 00:00:00 2001 From: peterdudfield Date: Wed, 31 Jul 2024 15:31:45 +0100 Subject: [PATCH] update test, remove try and except --- src/national.py | 78 ++++++++++----------- src/tests/test_elexon_forecast.py | 112 +++++++++++++++--------------- 2 files changed, 92 insertions(+), 98 deletions(-) diff --git a/src/national.py b/src/national.py index 75156f5..9c53956 100644 --- a/src/national.py +++ b/src/national.py @@ -41,6 +41,7 @@ # Initialize Elexon API client api_client = ApiClient() forecast_api = GenerationForecastApi(api_client) +forecast_generation_wind_and_solar_day_ahead_get = forecast_api.forecast_generation_wind_and_solar_day_ahead_get @router.get("/elexon", summary="Get elexon Solar Forecast") @@ -67,50 +68,45 @@ def get_elexon_forecast( Returns: SolarForecastResponse: The forecast data wrapped in a SolarForecastResponse model. """ - try: - response = forecast_api.forecast_generation_wind_and_solar_day_ahead_get( - _from=start_datetime_utc.isoformat(), - to=end_datetime_utc.isoformat(), - process_type=process_type, - format="json", - ) - if not response.data: - return SolarForecastResponse(data=[]) - - df = pd.DataFrame([item.to_dict() for item in response.data]) - logger.debug("DataFrame Columns: %s", df.columns) - logger.debug("DataFrame Sample: %s", df.head()) - - # Filter to include only solar forecasts - solar_df = df[df["business_type"] == "Solar generation"] - logger.debug("Filtered Solar DataFrame: %s", solar_df.head()) - - forecast_values = [] - for _, row in solar_df.iterrows(): - try: - forecast_values.append( - SolarForecastValue( - timestamp=pd.to_datetime(row["publish_time"]).to_pydatetime(), - expected_power_generation_megawatts=row.get("quantity"), - plevels=None, - ) - ) - except KeyError as e: - logger.error("KeyError: %s. Data: %s", str(e), row) - raise HTTPException(status_code=500, detail="Internal Server Error") - except Exception as e: - logger.error( - "Error during DataFrame to Model conversion: %s. Data: %s", str(e), row - ) - raise HTTPException(status_code=500, detail="Internal Server Error") + response = forecast_generation_wind_and_solar_day_ahead_get( + _from=start_datetime_utc.isoformat(), + to=end_datetime_utc.isoformat(), + process_type=process_type, + format="json", + ) + + if not response.data: + return SolarForecastResponse(data=[]) - result = SolarForecastResponse(data=forecast_values) - return result + df = pd.DataFrame([item.to_dict() for item in response.data]) + logger.debug("DataFrame Columns: %s", df.columns) + logger.debug("DataFrame Sample: %s", df.head()) + + # Filter to include only solar forecasts + solar_df = df[df["business_type"] == "Solar generation"] + logger.debug("Filtered Solar DataFrame: %s", solar_df.head()) + + forecast_values = [] + for _, row in solar_df.iterrows(): + try: + forecast_values.append( + SolarForecastValue( + timestamp=pd.to_datetime(row["publish_time"]).to_pydatetime(), + expected_power_generation_megawatts=row.get("quantity"), + ) + ) + except KeyError as e: + logger.error("KeyError: %s. Data: %s", str(e), row) + raise HTTPException(status_code=500, detail="Internal Server Error") + except Exception as e: + logger.error( + "Error during DataFrame to Model conversion: %s. Data: %s", str(e), row + ) + raise HTTPException(status_code=500, detail="Internal Server Error") - except Exception as e: - logger.error("Unhandled exception: %s", str(e)) - raise HTTPException(status_code=500, detail="Internal Server Error") + result = SolarForecastResponse(data=forecast_values) + return result @router.get( diff --git a/src/tests/test_elexon_forecast.py b/src/tests/test_elexon_forecast.py index 52422d4..96d6ba2 100644 --- a/src/tests/test_elexon_forecast.py +++ b/src/tests/test_elexon_forecast.py @@ -1,61 +1,59 @@ -import pytest -import requests -import requests_mock +from unittest.mock import patch + +import pandas as pd + +from pydantic_models import BaseModel +from typing import Optional + +# InsightsApiModelsResponsesResponseWithMetadata1InsightsApiModelsResponsesTransparencyDayAheadGenerationForWindAndSolar API_URL = "/v0/solar/GB/national/elexon" -@pytest.fixture -def mock_data(): - return { - "data": [ - { - "timestamp": "2024-07-24T16:45:09+00:00", - "expected_power_generation_megawatts": 0, - "plevels": None, - }, - { - "timestamp": "2024-07-24T16:45:09+00:00", - "expected_power_generation_megawatts": 0, - "plevels": None, - }, - ] - } - - -def test_get_elexon_forecast_with_data(mock_data, api_client): - with requests_mock.Mocker() as m: - url = ( - f"https://data.elexon.co.uk/bmrs/api/v1/forecast/generation/wind-and-solar/day-ahead" - ) - m.get(url, json=mock_data, headers={"Content-Type": "application/json"}) - - response = api_client.get('/v0/solar/GB/national/elexon') - print("Response Headers:", response.headers) - # Assertions - assert response.status_code == 200 - assert response.headers.get("Content-Type") == "application/json" - assert response.json() == mock_data - - -@pytest.fixture -def empty_mock_data(): - return {"data": []} - - -def test_get_elexon_forecast_no_data(empty_mock_data): - with requests_mock.Mocker() as m: - url = ( - f"{API_URL}?start_datetime_utc=2024-07-22T10:56:59.194610" - f"&end_datetime_utc=2024-07-28T10:56:59.194680" - f"&process_type=Day Ahead" - ) - - m.get(url, json=empty_mock_data, headers={"Content-Type": "application/json"}) - - response = requests.get(url) - print("Response Headers:", response.headers) - # Assertions - assert response.status_code == 200 - assert response.headers.get("Content-Type") == "application/json" - assert response.json() == empty_mock_data +class MockClass(BaseModel): + + publish_time: str + quantity: float + business_type: Optional[str] = "Solar generation" + + def to_dict(self): + return self.__dict__ + + +mock_data = [ + MockClass( + **{ + "publish_time": "2024-07-24T16:45:09+00:00", + "quantity": 0, + } + ), + MockClass( + **{ + "publish_time": "2024-07-24T16:45:09+00:00", + "quantity": 0, + } + ), +] + + +class MockResponse: + def __init__(self): + self.data = mock_data + + +@patch("national.forecast_generation_wind_and_solar_day_ahead_get") +def test_get_elexon_forecast_with_data(mock_function, api_client): + mock_function.return_value = MockResponse() + + response = api_client.get("/v0/solar/GB/national/elexon") + print("Response Headers:", response.headers) + # Assertions + assert response.status_code == 200 + assert response.headers.get("Content-Type") == "application/json" + + api_data = response.json()["data"] + assert len(api_data) == len(mock_data) + for i in range(len(api_data)): + assert api_data[i]["expected_power_generation_megawatts"] == mock_data[i].quantity + assert pd.Timestamp(api_data[i]["timestamp"]) == pd.Timestamp(mock_data[i].publish_time) +