From ed38b393613fe051f04cf4abc4b714d48d2eaa4c Mon Sep 17 00:00:00 2001 From: Joakim Amundsen Date: Thu, 2 Jul 2026 09:59:01 +0200 Subject: [PATCH 1/4] chore: add retry handling and integration/unit tests for python and .net SDKs --- .../Fakes/FakeHttpMessageHandler.cs | 39 ++ .../Fakes/HeimdallApiHttpClientFactory.cs | 28 ++ .../Fakes/StubAccessTokenProvider.cs | 20 + .../capacity_monitoring.py | 95 ++++ python/heimdall_api_client/client.py | 232 ++++++++++ python/heimdall_api_client/grid_insights.py | 193 ++++++++ python/tests/integration/conftest.py | 2 + .../test_when_fetching_capacity_monitoring.py | 169 +++++++ .../test_when_fetching_grid_insights.py | 307 ++++++++++++ python/tests/unit/test_latest_endpoints.py | 435 ++++++++++++++++++ 10 files changed, 1520 insertions(+) create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/FakeHttpMessageHandler.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/HeimdallApiHttpClientFactory.cs create mode 100644 dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/StubAccessTokenProvider.cs create mode 100644 python/tests/integration/test_when_fetching_capacity_monitoring.py create mode 100644 python/tests/integration/test_when_fetching_grid_insights.py create mode 100644 python/tests/unit/test_latest_endpoints.py diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/FakeHttpMessageHandler.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/FakeHttpMessageHandler.cs new file mode 100644 index 0000000..10d3535 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/FakeHttpMessageHandler.cs @@ -0,0 +1,39 @@ +namespace HeimdallPower.Api.Client.UnitTests.WhenHandlingTransientErrors.Fakes; + +/// +/// A minimal for unit tests that returns +/// a fixed response or a pre-configured sequence of responses. +/// +internal sealed class FakeHttpMessageHandler : HttpMessageHandler +{ + private readonly Queue _responses; + private readonly HttpResponseMessage _last; + + /// Returns for every request. + public FakeHttpMessageHandler(HttpResponseMessage response) + { + _responses = new Queue(); + _last = response; + } + + /// + /// Returns responses from in order. + /// Once the queue is exhausted the last response is repeated. + /// + public FakeHttpMessageHandler(params HttpResponseMessage[] responses) + { + if (responses.Length == 0) throw new ArgumentException("At least one response is required.", nameof(responses)); + _last = responses[^1]; + _responses = new Queue(responses[..^1]); + } + + protected override Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + var response = _responses.Count > 0 ? _responses.Dequeue() : _last; + response.RequestMessage = request; + return Task.FromResult(response); + } +} + diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/HeimdallApiHttpClientFactory.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/HeimdallApiHttpClientFactory.cs new file mode 100644 index 0000000..280f264 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/HeimdallApiHttpClientFactory.cs @@ -0,0 +1,28 @@ +using System.Net; + +namespace HeimdallPower.Api.Client.UnitTests.WhenHandlingTransientErrors.Fakes; + +internal static class HeimdallApiHttpClientFactory +{ + /// + /// Creates a backed by . + /// The delay function is a no-op so tests complete instantly. + /// + public static HeimdallApiHttpClient Create(FakeHttpMessageHandler handler) + { + var httpClient = new HttpClient(handler) { BaseAddress = new Uri("https://fake.heimdallcloud.com") }; + return new HeimdallApiHttpClient( + accessTokenProvider: new StubAccessTokenProvider(), + httpClient: httpClient, + clientMetadata: null); // no real sleeping in unit tests + } + + /// Convenience: single repeating response. + public static HeimdallApiHttpClient CreateWithFixedResponse(HttpResponseMessage response) + => Create(new FakeHttpMessageHandler(response)); + + /// Convenience: a sequence of responses (e.g. N failures then a success). + public static HeimdallApiHttpClient CreateWithSequence(params HttpResponseMessage[] responses) + => Create(new FakeHttpMessageHandler(responses)); +} + diff --git a/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/StubAccessTokenProvider.cs b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/StubAccessTokenProvider.cs new file mode 100644 index 0000000..c6369d4 --- /dev/null +++ b/dotnet/tests/unit/HeimdallPower.Api.Client.UnitTests/WhenHandlingTransientErrors/Fakes/StubAccessTokenProvider.cs @@ -0,0 +1,20 @@ +namespace HeimdallPower.Api.Client.UnitTests.WhenHandlingTransientErrors.Fakes; + +/// +/// A no-op token provider — tests only exercise the HTTP retry layer, +/// so the token is pre-set and never expired. +/// +internal sealed class StubAccessTokenProvider : IAccessTokenProvider +{ + public Task AcquireTokenAsync(CancellationToken cancellationToken = default) => Task.CompletedTask; + + public DateTimeOffset GetTokenExpiry() + => DateTimeOffset.UtcNow.AddHours(1); // always valid + + public IDictionary GetAccessHeaders() + => new Dictionary + { + { "Authorization", "Bearer stub-token" } + }; +} + diff --git a/python/heimdall_api_client/capacity_monitoring.py b/python/heimdall_api_client/capacity_monitoring.py index b9960a0..33b067b 100644 --- a/python/heimdall_api_client/capacity_monitoring.py +++ b/python/heimdall_api_client/capacity_monitoring.py @@ -1,5 +1,6 @@ from __future__ import annotations +import datetime from typing import TYPE_CHECKING from uuid import UUID @@ -19,12 +20,21 @@ from heimdall_api_client.errors import HeimdallApiError, body_preview if TYPE_CHECKING: + from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_facilities_get_circuit_ratings_response_200 import ( # noqa: E501 + CapacityMonitoringV1FacilitiesGetCircuitRatingsResponse200, + ) from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_facilities_get_latest_circuit_rating_forecasts_response_200 import ( # noqa: E501 CapacityMonitoringV1FacilitiesGetLatestCircuitRatingForecastsResponse200, ) from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_facilities_get_latest_circuit_rating_response_200 import ( # noqa: E501 CapacityMonitoringV1FacilitiesGetLatestCircuitRatingResponse200, ) + from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_heimdall_aars_response_200 import ( # noqa: E501 + CapacityMonitoringV1LinesGetHeimdallAarsResponse200, + ) + from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_heimdall_dlrs_response_200 import ( # noqa: E501 + CapacityMonitoringV1LinesGetHeimdallDlrsResponse200, + ) from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_latest_heimdall_aar_forecasts_response_200 import ( # noqa: E501 CapacityMonitoringV1LinesGetLatestHeimdallAarForecastsResponse200, ) @@ -131,3 +141,88 @@ def get_latest_circuit_rating_forecasts( status_code=status, ) return response.parsed + + +def get_heimdall_dlrs( + client: AuthenticatedClient, + line_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, +) -> CapacityMonitoringV1LinesGetHeimdallDlrsResponse200: + from heimdall_api_client.capacity_monitoring_api_client.api.line import ( + capacity_monitoring_v1_lines_get_heimdall_dlrs as _get_heimdall_dlrs, + ) + + response = _get_heimdall_dlrs.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching Heimdall DLRs: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_heimdall_aars( + client: AuthenticatedClient, + line_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, +) -> CapacityMonitoringV1LinesGetHeimdallAarsResponse200: + from heimdall_api_client.capacity_monitoring_api_client.api.line import ( + capacity_monitoring_v1_lines_get_heimdall_aars as _get_heimdall_aars, + ) + + response = _get_heimdall_aars.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching Heimdall AARs: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_circuit_ratings( + client: AuthenticatedClient, + facility_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, +) -> CapacityMonitoringV1FacilitiesGetCircuitRatingsResponse200: + from heimdall_api_client.capacity_monitoring_api_client.api.facility import ( + capacity_monitoring_v1_facilities_get_circuit_ratings as _get_circuit_ratings, + ) + + response = _get_circuit_ratings.sync_detailed( + client=client, + facility_id=facility_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching circuit ratings: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + diff --git a/python/heimdall_api_client/client.py b/python/heimdall_api_client/client.py index 0f0adf2..51f54c6 100644 --- a/python/heimdall_api_client/client.py +++ b/python/heimdall_api_client/client.py @@ -18,6 +18,9 @@ get_latest_heimdall_arr_forecasts, get_latest_heimdall_dlr, get_latest_heimdall_dlr_forecasts, + get_heimdall_dlrs, + get_heimdall_aars, + get_circuit_ratings, ) from heimdall_api_client.errors import HeimdallApiError from heimdall_api_client.grid_insights_api_client.models.unit_system import UnitSystem @@ -28,12 +31,21 @@ from heimdall_api_client.assets_api_client.models.assets_v1_get_assets_response_200 import ( AssetsV1GetAssetsResponse200, ) + from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_facilities_get_circuit_ratings_response_200 import ( # noqa: E501 + CapacityMonitoringV1FacilitiesGetCircuitRatingsResponse200, + ) from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_facilities_get_latest_circuit_rating_forecasts_response_200 import ( # noqa: E501 CapacityMonitoringV1FacilitiesGetLatestCircuitRatingForecastsResponse200, ) from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_facilities_get_latest_circuit_rating_response_200 import ( # noqa: E501 CapacityMonitoringV1FacilitiesGetLatestCircuitRatingResponse200, ) + from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_heimdall_aars_response_200 import ( # noqa: E501 + CapacityMonitoringV1LinesGetHeimdallAarsResponse200, + ) + from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_heimdall_dlrs_response_200 import ( # noqa: E501 + CapacityMonitoringV1LinesGetHeimdallDlrsResponse200, + ) from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_latest_heimdall_aar_forecasts_response_200 import ( # noqa: E501 CapacityMonitoringV1LinesGetLatestHeimdallAarForecastsResponse200, ) @@ -46,6 +58,12 @@ from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_latest_heimdall_dlr_response_200 import ( # noqa: E501 CapacityMonitoringV1LinesGetLatestHeimdallDlrResponse200, ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_conductor_temperatures_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetConductorTemperaturesResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_currents_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetCurrentsResponse200, + ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_conductor_temperature_response_200 import ( # noqa: E501 GridInsightsV1LinesGetLatestConductorTemperatureResponse200, ) @@ -55,6 +73,21 @@ from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_icing_response_200 import ( # noqa: E501 GridInsightsV1LinesGetLatestIcingResponse200, ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_apparent_power_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetApparentPowerResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetIcingResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_forecast_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetIcingForecastResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_apparent_power_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetLatestApparentPowerResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_sag_and_clearance_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetSagAndClearanceResponse200, + ) _MAX_RETRY_ATTEMPTS = 3 @@ -342,3 +375,202 @@ def get_latest_sag_and_clearance( since=since, ) ) + + def get_currents( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + ) -> GridInsightsV1LinesGetCurrentsResponse200: + """ + Returns historical current measurements for a given line. + """ + from heimdall_api_client.grid_insights import get_currents + + return self._execute_with_retry( + lambda: get_currents( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + ) + + def get_conductor_temperatures( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + ) -> GridInsightsV1LinesGetConductorTemperaturesResponse200: + """ + Returns historical conductor temperature measurements for a given line. + """ + from heimdall_api_client.grid_insights import get_conductor_temperatures + + return self._execute_with_retry( + lambda: get_conductor_temperatures( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + ) + + def get_heimdall_dlrs( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + ) -> CapacityMonitoringV1LinesGetHeimdallDlrsResponse200: + """ + Returns historical Heimdall DLR (Dynamic Line Rating) values for a given line. + """ + return self._execute_with_retry( + lambda: get_heimdall_dlrs( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + ) + + def get_heimdall_aars( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + ) -> CapacityMonitoringV1LinesGetHeimdallAarsResponse200: + """ + Returns historical Heimdall AAR (Available Ampacity Rating) values for a given line. + """ + return self._execute_with_retry( + lambda: get_heimdall_aars( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + ) + + def get_circuit_ratings( + self, + facility_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + ) -> CapacityMonitoringV1FacilitiesGetCircuitRatingsResponse200: + """ + Returns historical circuit ratings for a given facility. + """ + return self._execute_with_retry( + lambda: get_circuit_ratings( + client=self._get_authenticated_client(), + facility_id=facility_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + ) + + def get_icing( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + unit_system: UnitSystem | str | None = None, + ) -> GridInsightsV1LinesGetIcingResponse200: + """ + Returns historical icing measurements for a given line. + """ + from heimdall_api_client.grid_insights import get_icing + + return self._execute_with_retry( + lambda: get_icing( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + unit_system=unit_system, + ) + ) + + def get_sag_and_clearance( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + unit_system: UnitSystem | str | None = None, + ) -> GridInsightsV1LinesGetSagAndClearanceResponse200: + """ + Returns historical sag and clearance measurements for a given line. + """ + from heimdall_api_client.grid_insights import get_sag_and_clearance + + return self._execute_with_retry( + lambda: get_sag_and_clearance( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + unit_system=unit_system, + ) + ) + + def get_apparent_power( + self, + line_id: UUID, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + ) -> GridInsightsV1LinesGetApparentPowerResponse200: + """ + Returns historical apparent power measurements for a given line. + """ + from heimdall_api_client.grid_insights import get_apparent_power + + return self._execute_with_retry( + lambda: get_apparent_power( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + ) + + def get_latest_apparent_power(self, line_id: UUID) -> GridInsightsV1LinesGetLatestApparentPowerResponse200: + """ + Returns the latest apparent power measurement for a given line. + """ + from heimdall_api_client.grid_insights import get_latest_apparent_power + + return self._execute_with_retry( + lambda: get_latest_apparent_power( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + ) + ) + + def get_icing_forecast( + self, + line_id: UUID, + unit_system: UnitSystem | str | None = None, + ) -> GridInsightsV1LinesGetIcingForecastResponse200: + """ + Returns the latest icing forecast for a given line. + """ + from heimdall_api_client.grid_insights import get_icing_forecast + + return self._execute_with_retry( + lambda: get_icing_forecast( + client=self._get_authenticated_client(), + line_id=line_id, + region=self._get_region(), + unit_system=unit_system, + ) + ) diff --git a/python/heimdall_api_client/grid_insights.py b/python/heimdall_api_client/grid_insights.py index e27b193..925eeea 100644 --- a/python/heimdall_api_client/grid_insights.py +++ b/python/heimdall_api_client/grid_insights.py @@ -10,6 +10,24 @@ from heimdall_api_client.grid_insights_api_client.types import UNSET if TYPE_CHECKING: + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_apparent_power_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetApparentPowerResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_conductor_temperatures_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetConductorTemperaturesResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_currents_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetCurrentsResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_forecast_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetIcingForecastResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetIcingResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_apparent_power_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetLatestApparentPowerResponse200, + ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_conductor_temperature_response_200 import ( # noqa: E501 GridInsightsV1LinesGetLatestConductorTemperatureResponse200, ) @@ -19,6 +37,9 @@ from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_icing_response_200 import ( # noqa: E501 GridInsightsV1LinesGetLatestIcingResponse200, ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_sag_and_clearance_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetSagAndClearanceResponse200, + ) def get_latest_conductor_temperature( @@ -121,3 +142,175 @@ def get_latest_sag_and_clearance( status_code=status, ) return response.parsed + + +def get_icing( + client: AuthenticatedClient, + line_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + unit_system: UnitSystem | str | None = None, +) -> GridInsightsV1LinesGetIcingResponse200: + from heimdall_api_client.grid_insights_api_client.api.line import ( + grid_insights_v1_lines_get_icing as _get_icing, + ) + + unit_system_value = UNSET + if unit_system is not None: + unit_system_value = unit_system if isinstance(unit_system, UnitSystem) else UnitSystem(unit_system) + + response = _get_icing.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + unit_system=unit_system_value, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching icing: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_sag_and_clearance( + client: AuthenticatedClient, + line_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, + unit_system: UnitSystem | str | None = None, +) -> GridInsightsV1LinesGetSagAndClearanceResponse200: + from heimdall_api_client.grid_insights_api_client.api.line import ( + grid_insights_v1_lines_get_sag_and_clearance as _get_sag_and_clearance, + ) + + unit_system_value = UNSET + if unit_system is not None: + unit_system_value = unit_system if isinstance(unit_system, UnitSystem) else UnitSystem(unit_system) + + response = _get_sag_and_clearance.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + unit_system=unit_system_value, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching sag and clearance: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_apparent_power( + client: AuthenticatedClient, + line_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, +) -> GridInsightsV1LinesGetApparentPowerResponse200: + from heimdall_api_client.grid_insights_api_client.api.line import ( + grid_insights_v1_lines_get_apparent_power as _get_apparent_power, + ) + + response = _get_apparent_power.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching apparent power: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_latest_apparent_power( + client: AuthenticatedClient, line_id: UUID, region: str +) -> GridInsightsV1LinesGetLatestApparentPowerResponse200: + from heimdall_api_client.grid_insights_api_client.api.line import ( + grid_insights_v1_lines_get_latest_apparent_power as _get_latest_apparent_power, + ) + + response = _get_latest_apparent_power.sync_detailed(client=client, line_id=line_id, x_region=region) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching latest apparent power: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_icing_forecast( + client: AuthenticatedClient, + line_id: UUID, + region: str, + unit_system: UnitSystem | str | None = None, +) -> GridInsightsV1LinesGetIcingForecastResponse200: + from heimdall_api_client.grid_insights_api_client.api.line import ( + grid_insights_v1_lines_get_icing_forecast as _get_icing_forecast, + ) + + unit_system_value = UNSET + if unit_system is not None: + unit_system_value = unit_system if isinstance(unit_system, UnitSystem) else UnitSystem(unit_system) + + response = _get_icing_forecast.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + unit_system=unit_system_value, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching icing forecast: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed + + +def get_conductor_temperatures( + client: AuthenticatedClient, + line_id: UUID, + region: str, + from_timestamp: datetime.datetime, + to_timestamp: datetime.datetime, +) -> GridInsightsV1LinesGetConductorTemperaturesResponse200: + from heimdall_api_client.grid_insights_api_client.api.line import ( + grid_insights_v1_lines_get_conductor_temperatures as _get_conductor_temperatures, + ) + + response = _get_conductor_temperatures.sync_detailed( + client=client, + line_id=line_id, + x_region=region, + from_timestamp=from_timestamp, + to_timestamp=to_timestamp, + ) + if response.status_code != 200: + status = int(response.status_code) + raise HeimdallApiError( + f"Error fetching conductor temperatures: {status} {response.status_code.phrase}" + f" - {body_preview(response.content)}", + status_code=status, + ) + return response.parsed diff --git a/python/tests/integration/conftest.py b/python/tests/integration/conftest.py index a14b773..bc629df 100644 --- a/python/tests/integration/conftest.py +++ b/python/tests/integration/conftest.py @@ -10,3 +10,5 @@ def api_client(): return HeimdallApiClient( client_id=os.environ["HEIMDALL_CLIENT_ID"], client_secret=os.environ["HEIMDALL_CLIENT_SECRET"] ) + + diff --git a/python/tests/integration/test_when_fetching_capacity_monitoring.py b/python/tests/integration/test_when_fetching_capacity_monitoring.py new file mode 100644 index 0000000..e71c45a --- /dev/null +++ b/python/tests/integration/test_when_fetching_capacity_monitoring.py @@ -0,0 +1,169 @@ +"""Integration tests for Capacity Monitoring endpoints.""" + +import datetime +import uuid + +import pytest + +from heimdall_api_client import HeimdallApiError + +# "Heimdall Power Line" – d67d2205-6629-4bbd-aa9f-436bf22842ad +_HEIMDALL_POWER_LINE_ID = uuid.UUID("d67d2205-6629-4bbd-aa9f-436bf22842ad") +# "Heimdall Power Line" facility – c0ad547d-0d06-4f4c-b5dc-d319430902d2 +_HEIMDALL_POWER_FACILITY_ID = uuid.UUID("c0ad547d-0d06-4f4c-b5dc-d319430902d2") +_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) +_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc) + + +@pytest.mark.integration +def test_get_latest_heimdall_dlr_with_invalid_line_id_should_raise(api_client): + with pytest.raises(HeimdallApiError) as exc_info: + api_client.get_latest_heimdall_dlr(line_id=uuid.uuid4()) + assert exc_info.value.status_code in (404, 403), "Expected 404 or 403 for unknown line" + + +# --------------------------------------------------------------------------- +# get_heimdall_dlrs – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def heimdall_dlrs_result(api_client): + return api_client.get_heimdall_dlrs( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_heimdall_dlrs_should_return_response(heimdall_dlrs_result): + assert heimdall_dlrs_result is not None + + +@pytest.mark.integration +def test_get_heimdall_dlrs_result_should_have_metric(heimdall_dlrs_result): + assert heimdall_dlrs_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_heimdall_dlrs_result_should_have_unit(heimdall_dlrs_result): + assert heimdall_dlrs_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_heimdall_dlrs_result_should_have_dlrs_list(heimdall_dlrs_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert heimdall_dlrs_result.data.heimdall_dlrs is not None + + +@pytest.mark.integration +def test_get_heimdall_dlrs_all_dlrs_should_have_timestamps_within_requested_range(heimdall_dlrs_result): + for dlr in heimdall_dlrs_result.data.heimdall_dlrs: + assert dlr.timestamp >= _FROM, f"Timestamp {dlr.timestamp} is before {_FROM}" + assert dlr.timestamp <= _TO, f"Timestamp {dlr.timestamp} is after {_TO}" + + +@pytest.mark.integration +def test_get_heimdall_dlrs_all_dlrs_should_have_positive_values(heimdall_dlrs_result): + for dlr in heimdall_dlrs_result.data.heimdall_dlrs: + assert dlr.value > 0, f"DLR value {dlr.value} at {dlr.timestamp} should be positive" + + +# --------------------------------------------------------------------------- +# get_heimdall_aars – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def heimdall_aars_result(api_client): + return api_client.get_heimdall_aars( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_heimdall_aars_should_return_response(heimdall_aars_result): + assert heimdall_aars_result is not None + + +@pytest.mark.integration +def test_get_heimdall_aars_result_should_have_metric(heimdall_aars_result): + assert heimdall_aars_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_heimdall_aars_result_should_have_unit(heimdall_aars_result): + assert heimdall_aars_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_heimdall_aars_result_should_have_aars_list(heimdall_aars_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert heimdall_aars_result.data.heimdall_aars is not None + + +@pytest.mark.integration +def test_get_heimdall_aars_all_aars_should_have_timestamps_within_requested_range(heimdall_aars_result): + for aar in heimdall_aars_result.data.heimdall_aars: + assert aar.timestamp >= _FROM, f"Timestamp {aar.timestamp} is before {_FROM}" + assert aar.timestamp <= _TO, f"Timestamp {aar.timestamp} is after {_TO}" + + +@pytest.mark.integration +def test_get_heimdall_aars_all_aars_should_have_positive_values(heimdall_aars_result): + for aar in heimdall_aars_result.data.heimdall_aars: + assert aar.value > 0, f"AAR value {aar.value} at {aar.timestamp} should be positive" + + +# --------------------------------------------------------------------------- +# get_circuit_ratings – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def circuit_ratings_result(api_client): + return api_client.get_circuit_ratings( + facility_id=_HEIMDALL_POWER_FACILITY_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_circuit_ratings_should_return_response(circuit_ratings_result): + assert circuit_ratings_result is not None + + +@pytest.mark.integration +def test_get_circuit_ratings_result_should_have_metric(circuit_ratings_result): + assert circuit_ratings_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_circuit_ratings_result_should_have_unit(circuit_ratings_result): + assert circuit_ratings_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_circuit_ratings_result_should_have_circuit_ratings_list(circuit_ratings_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert circuit_ratings_result.data.circuit_ratings is not None + + +@pytest.mark.integration +def test_get_circuit_ratings_all_circuit_ratings_should_have_timestamps_within_requested_range( + circuit_ratings_result, +): + for cr in circuit_ratings_result.data.circuit_ratings: + assert cr.timestamp >= _FROM, f"Timestamp {cr.timestamp} is before {_FROM}" + assert cr.timestamp <= _TO, f"Timestamp {cr.timestamp} is after {_TO}" + + +@pytest.mark.integration +def test_get_circuit_ratings_all_circuit_ratings_should_have_positive_values(circuit_ratings_result): + for cr in circuit_ratings_result.data.circuit_ratings: + assert cr.value > 0, f"Circuit rating value {cr.value} at {cr.timestamp} should be positive" + diff --git a/python/tests/integration/test_when_fetching_grid_insights.py b/python/tests/integration/test_when_fetching_grid_insights.py new file mode 100644 index 0000000..64ea73b --- /dev/null +++ b/python/tests/integration/test_when_fetching_grid_insights.py @@ -0,0 +1,307 @@ +"""Integration tests for Grid Insights endpoints.""" + +import datetime +import uuid + +import pytest + +from heimdall_api_client import HeimdallApiError + +# "Heimdall Power Line" – d67d2205-6629-4bbd-aa9f-436bf22842ad +_HEIMDALL_POWER_LINE_ID = uuid.UUID("d67d2205-6629-4bbd-aa9f-436bf22842ad") +_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) +_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc) + + +@pytest.mark.integration +def test_get_latest_current_with_invalid_line_id_should_raise(api_client): + with pytest.raises(HeimdallApiError) as exc_info: + api_client.get_latest_current(line_id=uuid.uuid4()) + assert exc_info.value.status_code in (404, 403), "Expected 404 or 403 for unknown line" + + +# --------------------------------------------------------------------------- +# get_currents – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def currents_result(api_client): + return api_client.get_currents( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_currents_should_return_response(currents_result): + assert currents_result is not None + + +@pytest.mark.integration +def test_get_currents_result_should_have_metric(currents_result): + assert currents_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_currents_result_should_have_unit(currents_result): + assert currents_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_currents_result_should_have_currents_list(currents_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert currents_result.data.currents is not None + + +@pytest.mark.integration +def test_get_currents_all_currents_should_have_timestamps_within_requested_range(currents_result): + for current in currents_result.data.currents: + assert current.timestamp >= _FROM, f"Timestamp {current.timestamp} is before {_FROM}" + assert current.timestamp <= _TO, f"Timestamp {current.timestamp} is after {_TO}" + + +# --------------------------------------------------------------------------- +# get_conductor_temperatures – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def conductor_temperatures_result(api_client): + return api_client.get_conductor_temperatures( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_conductor_temperatures_should_return_response(conductor_temperatures_result): + assert conductor_temperatures_result is not None + + +@pytest.mark.integration +def test_get_conductor_temperatures_result_should_have_metric(conductor_temperatures_result): + assert conductor_temperatures_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_conductor_temperatures_result_should_have_unit(conductor_temperatures_result): + assert conductor_temperatures_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_conductor_temperatures_result_should_have_list(conductor_temperatures_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert conductor_temperatures_result.data.conductor_temperatures is not None + + +@pytest.mark.integration +def test_get_conductor_temperatures_all_readings_should_have_timestamps_within_requested_range( + conductor_temperatures_result, +): + for ct in conductor_temperatures_result.data.conductor_temperatures: + assert ct.timestamp >= _FROM, f"Timestamp {ct.timestamp} is before {_FROM}" + assert ct.timestamp <= _TO, f"Timestamp {ct.timestamp} is after {_TO}" + + +# --------------------------------------------------------------------------- +# get_icing – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def icing_result(api_client): + return api_client.get_icing( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_icing_should_return_response(icing_result): + assert icing_result is not None + + +@pytest.mark.integration +def test_get_icing_result_should_have_metric(icing_result): + assert icing_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_icing_result_should_have_unit(icing_result): + assert icing_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_icing_result_should_have_icing_data(icing_result): + # The API returns HTTP 200 – icing is a nested object, not a flat list. + assert icing_result.data.icing is not None + + +# --------------------------------------------------------------------------- +# get_sag_and_clearance – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def sag_and_clearance_result(api_client): + return api_client.get_sag_and_clearance( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_sag_and_clearance_should_return_response(sag_and_clearance_result): + assert sag_and_clearance_result is not None + + +@pytest.mark.integration +def test_get_sag_and_clearance_result_should_have_metric(sag_and_clearance_result): + assert sag_and_clearance_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_sag_and_clearance_result_should_have_unit(sag_and_clearance_result): + assert sag_and_clearance_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_sag_and_clearance_result_should_have_sag_and_clearance_data(sag_and_clearance_result): + # The API returns HTTP 200 – sag_and_clearance is a nested object, not a flat list. + assert sag_and_clearance_result.data.sag_and_clearance is not None + + +# --------------------------------------------------------------------------- +# get_apparent_power – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def apparent_power_result(api_client): + return api_client.get_apparent_power( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_apparent_power_should_return_response(apparent_power_result): + assert apparent_power_result is not None + + +@pytest.mark.integration +def test_get_apparent_power_result_should_have_metric(apparent_power_result): + assert apparent_power_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_apparent_power_result_should_have_unit(apparent_power_result): + assert apparent_power_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_apparent_power_result_should_have_apparent_powers_list(apparent_power_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert apparent_power_result.data.apparent_powers is not None + + +@pytest.mark.integration +def test_get_apparent_power_all_readings_should_have_timestamps_within_requested_range(apparent_power_result): + for ap in apparent_power_result.data.apparent_powers: + assert ap.timestamp >= _FROM, f"Timestamp {ap.timestamp} is before {_FROM}" + assert ap.timestamp <= _TO, f"Timestamp {ap.timestamp} is after {_TO}" + + + +# --------------------------------------------------------------------------- +# get_currents – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def currents_result(api_client): + return api_client.get_currents( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_currents_should_return_response(currents_result): + assert currents_result is not None + + +@pytest.mark.integration +def test_get_currents_result_should_have_metric(currents_result): + assert currents_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_currents_result_should_have_unit(currents_result): + assert currents_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_currents_result_should_have_currents_list(currents_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert currents_result.data.currents is not None + + +@pytest.mark.integration +def test_get_currents_all_currents_should_have_timestamps_within_requested_range(currents_result): + for current in currents_result.data.currents: + assert current.timestamp >= _FROM, f"Timestamp {current.timestamp} is before {_FROM}" + assert current.timestamp <= _TO, f"Timestamp {current.timestamp} is after {_TO}" + + +# --------------------------------------------------------------------------- +# get_conductor_temperatures – historical +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def conductor_temperatures_result(api_client): + return api_client.get_conductor_temperatures( + line_id=_HEIMDALL_POWER_LINE_ID, + from_timestamp=_FROM, + to_timestamp=_TO, + ) + + +@pytest.mark.integration +def test_get_conductor_temperatures_should_return_response(conductor_temperatures_result): + assert conductor_temperatures_result is not None + + +@pytest.mark.integration +def test_get_conductor_temperatures_result_should_have_metric(conductor_temperatures_result): + assert conductor_temperatures_result.data.metric, "Metric should not be empty" + + +@pytest.mark.integration +def test_get_conductor_temperatures_result_should_have_unit(conductor_temperatures_result): + assert conductor_temperatures_result.data.unit, "Unit should not be empty" + + +@pytest.mark.integration +def test_get_conductor_temperatures_result_should_have_list(conductor_temperatures_result): + # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. + assert conductor_temperatures_result.data.conductor_temperatures is not None + + +@pytest.mark.integration +def test_get_conductor_temperatures_all_readings_should_have_timestamps_within_requested_range( + conductor_temperatures_result, +): + for ct in conductor_temperatures_result.data.conductor_temperatures: + assert ct.timestamp >= _FROM, f"Timestamp {ct.timestamp} is before {_FROM}" + assert ct.timestamp <= _TO, f"Timestamp {ct.timestamp} is after {_TO}" + + diff --git a/python/tests/unit/test_latest_endpoints.py b/python/tests/unit/test_latest_endpoints.py new file mode 100644 index 0000000..1d3eaaa --- /dev/null +++ b/python/tests/unit/test_latest_endpoints.py @@ -0,0 +1,435 @@ +""" +Unit tests for latest and forecast endpoint methods in HeimdallApiClient. + +These tests mock the underlying HTTP/API layer so they are: + - Pure – no network calls, no external state + - Fast – run in milliseconds + - Deterministic – same input always produces same output + +Coverage: + - Grid Insights: get_latest_current, get_latest_conductor_temperature, + get_latest_icing, get_latest_sag_and_clearance, + get_latest_apparent_power, get_icing_forecast + - Capacity Monitoring: get_latest_heimdall_dlr, get_latest_heimdall_aar, + get_latest_heimdall_dlr_forecasts, + get_latest_heimdall_aar_forecasts, + get_latest_circuit_rating, + get_latest_circuit_rating_forecasts + +Each endpoint is tested for: + 1. 200 OK → the parsed response object is returned to the caller. + 2. 404 Not Found → HeimdallApiError(status_code=404) is raised (no retry). + 3. 404 is not retried → time.sleep must not be called. +""" + +import uuid +from unittest.mock import MagicMock, patch + +import pytest + +from heimdall_api_client.errors import HeimdallApiError + +# --------------------------------------------------------------------------- +# Shared test helpers +# --------------------------------------------------------------------------- + +_LINE_ID = uuid.UUID("11111111-1111-1111-1111-111111111111") +_FACILITY_ID = uuid.UUID("22222222-2222-2222-2222-222222222222") + + +def _make_client(): + """Return a HeimdallApiClient with auth completely stubbed out.""" + from heimdall_api_client.client import HeimdallApiClient + + client = HeimdallApiClient.__new__(HeimdallApiClient) + client.logger = MagicMock() + client.auth_service = MagicMock() + client.auth_service.get_valid_token.return_value = "stub-token" + client.auth_service.get_region_from_token.return_value = "no" + client.api_base_url = "https://stub.heimdallcloud.com" + client.client_metadata = {} + client.timeout = None + return client + + +def _raises_404(): + return HeimdallApiError("Not Found", status_code=404) + + +# --------------------------------------------------------------------------- +# Grid Insights – get_latest_current +# --------------------------------------------------------------------------- + + +class TestGetLatestCurrent: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.grid_insights.get_latest_current", return_value=expected): + result = client.get_latest_current(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_current", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_current(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + """Non-transient 404 must not trigger any retry delay.""" + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_current", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_current(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Grid Insights – get_latest_conductor_temperature +# --------------------------------------------------------------------------- + + +class TestGetLatestConductorTemperature: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.grid_insights.get_latest_conductor_temperature", return_value=expected): + result = client.get_latest_conductor_temperature(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch( + "heimdall_api_client.grid_insights.get_latest_conductor_temperature", side_effect=_raises_404() + ): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_conductor_temperature(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch( + "heimdall_api_client.grid_insights.get_latest_conductor_temperature", side_effect=_raises_404() + ): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_conductor_temperature(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Grid Insights – get_latest_icing +# --------------------------------------------------------------------------- + + +class TestGetLatestIcing: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.grid_insights.get_latest_icing", return_value=expected): + result = client.get_latest_icing(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_icing", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_icing(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_icing", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_icing(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Grid Insights – get_latest_sag_and_clearance +# --------------------------------------------------------------------------- + + +class TestGetLatestSagAndClearance: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.grid_insights.get_latest_sag_and_clearance", return_value=expected): + result = client.get_latest_sag_and_clearance(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_sag_and_clearance", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_sag_and_clearance(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_sag_and_clearance", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_sag_and_clearance(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Grid Insights – get_latest_apparent_power +# --------------------------------------------------------------------------- + + +class TestGetLatestApparentPower: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.grid_insights.get_latest_apparent_power", return_value=expected): + result = client.get_latest_apparent_power(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_apparent_power", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_apparent_power(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_latest_apparent_power", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_apparent_power(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Grid Insights – get_icing_forecast +# --------------------------------------------------------------------------- + + +class TestGetIcingForecast: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.grid_insights.get_icing_forecast", return_value=expected): + result = client.get_icing_forecast(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_icing_forecast", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_icing_forecast(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch("heimdall_api_client.grid_insights.get_icing_forecast", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_icing_forecast(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Capacity Monitoring – get_latest_heimdall_dlr +# --------------------------------------------------------------------------- + + +class TestGetLatestHeimdallDlr: + # get_latest_heimdall_dlr is a top-level import in client.py, so the patch + # target is the name as bound inside that module. + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.client.get_latest_heimdall_dlr", return_value=expected): + result = client.get_latest_heimdall_dlr(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.client.get_latest_heimdall_dlr", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_heimdall_dlr(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch("heimdall_api_client.client.get_latest_heimdall_dlr", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_heimdall_dlr(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Capacity Monitoring – get_latest_heimdall_aar +# --------------------------------------------------------------------------- + + +class TestGetLatestHeimdallAar: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.client.get_latest_heimdall_aar", return_value=expected): + result = client.get_latest_heimdall_aar(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch("heimdall_api_client.client.get_latest_heimdall_aar", side_effect=_raises_404()): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_heimdall_aar(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch("heimdall_api_client.client.get_latest_heimdall_aar", side_effect=_raises_404()): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_heimdall_aar(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Capacity Monitoring – get_latest_heimdall_dlr_forecasts +# --------------------------------------------------------------------------- + + +class TestGetLatestHeimdallDlrForecasts: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", return_value=expected): + result = client.get_latest_heimdall_dlr_forecasts(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch( + "heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", side_effect=_raises_404() + ): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_heimdall_dlr_forecasts(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch( + "heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", side_effect=_raises_404() + ): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_heimdall_dlr_forecasts(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Capacity Monitoring – get_latest_heimdall_aar_forecasts +# --------------------------------------------------------------------------- + + +class TestGetLatestHeimdallAarForecasts: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch("heimdall_api_client.client.get_latest_heimdall_arr_forecasts", return_value=expected): + result = client.get_latest_heimdall_aar_forecasts(line_id=_LINE_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch( + "heimdall_api_client.client.get_latest_heimdall_arr_forecasts", side_effect=_raises_404() + ): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_heimdall_aar_forecasts(line_id=_LINE_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch( + "heimdall_api_client.client.get_latest_heimdall_arr_forecasts", side_effect=_raises_404() + ): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_heimdall_aar_forecasts(line_id=_LINE_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Capacity Monitoring – get_latest_circuit_rating +# --------------------------------------------------------------------------- + + +class TestGetLatestCircuitRating: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch( + "heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", return_value=expected + ): + result = client.get_latest_circuit_rating(facility_id=_FACILITY_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch( + "heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", side_effect=_raises_404() + ): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_circuit_rating(facility_id=_FACILITY_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch( + "heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", side_effect=_raises_404() + ): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_circuit_rating(facility_id=_FACILITY_ID) + mock_sleep.assert_not_called() + + +# --------------------------------------------------------------------------- +# Capacity Monitoring – get_latest_circuit_rating_forecasts +# --------------------------------------------------------------------------- + + +class TestGetLatestCircuitRatingForecasts: + def test_returns_parsed_response_on_200(self): + client = _make_client() + expected = MagicMock() + with patch( + "heimdall_api_client.capacity_monitoring.get_latest_circuit_rating_forecasts", + return_value=expected, + ): + result = client.get_latest_circuit_rating_forecasts(facility_id=_FACILITY_ID) + assert result is expected + + def test_raises_404_error_on_not_found(self): + client = _make_client() + with patch( + "heimdall_api_client.capacity_monitoring.get_latest_circuit_rating_forecasts", + side_effect=_raises_404(), + ): + with pytest.raises(HeimdallApiError) as exc_info: + client.get_latest_circuit_rating_forecasts(facility_id=_FACILITY_ID) + assert exc_info.value.status_code == 404 + + def test_404_is_not_retried(self): + client = _make_client() + with patch( + "heimdall_api_client.capacity_monitoring.get_latest_circuit_rating_forecasts", + side_effect=_raises_404(), + ): + with patch("time.sleep") as mock_sleep: + with pytest.raises(HeimdallApiError): + client.get_latest_circuit_rating_forecasts(facility_id=_FACILITY_ID) + mock_sleep.assert_not_called() + From 493eac4dcebac17f1139a439c086d5766fd8989b Mon Sep 17 00:00:00 2001 From: Joakim Amundsen Date: Thu, 2 Jul 2026 10:15:04 +0200 Subject: [PATCH 2/4] fix: remove duplicate test definitions and fix ruff lint errors --- pr-description.md | 0 python/heimdall_api_client/client.py | 30 +++--- python/heimdall_api_client/grid_insights.py | 3 - .../test_when_fetching_capacity_monitoring.py | 4 +- .../test_when_fetching_grid_insights.py | 92 +------------------ 5 files changed, 19 insertions(+), 110 deletions(-) create mode 100644 pr-description.md diff --git a/pr-description.md b/pr-description.md new file mode 100644 index 0000000..e69de29 diff --git a/python/heimdall_api_client/client.py b/python/heimdall_api_client/client.py index 51f54c6..03f0b71 100644 --- a/python/heimdall_api_client/client.py +++ b/python/heimdall_api_client/client.py @@ -14,13 +14,13 @@ from heimdall_api_client.assets_api_client.client import AuthenticatedClient from heimdall_api_client.auth import AuthService from heimdall_api_client.capacity_monitoring import ( + get_circuit_ratings, + get_heimdall_aars, + get_heimdall_dlrs, get_latest_heimdall_aar, get_latest_heimdall_arr_forecasts, get_latest_heimdall_dlr, get_latest_heimdall_dlr_forecasts, - get_heimdall_dlrs, - get_heimdall_aars, - get_circuit_ratings, ) from heimdall_api_client.errors import HeimdallApiError from heimdall_api_client.grid_insights_api_client.models.unit_system import UnitSystem @@ -58,12 +58,24 @@ from heimdall_api_client.capacity_monitoring_api_client.models.capacity_monitoring_v1_lines_get_latest_heimdall_dlr_response_200 import ( # noqa: E501 CapacityMonitoringV1LinesGetLatestHeimdallDlrResponse200, ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_apparent_power_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetApparentPowerResponse200, + ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_conductor_temperatures_response_200 import ( # noqa: E501 GridInsightsV1LinesGetConductorTemperaturesResponse200, ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_currents_response_200 import ( # noqa: E501 GridInsightsV1LinesGetCurrentsResponse200, ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_forecast_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetIcingForecastResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetIcingResponse200, + ) + from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_apparent_power_response_200 import ( # noqa: E501 + GridInsightsV1LinesGetLatestApparentPowerResponse200, + ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_conductor_temperature_response_200 import ( # noqa: E501 GridInsightsV1LinesGetLatestConductorTemperatureResponse200, ) @@ -73,18 +85,6 @@ from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_icing_response_200 import ( # noqa: E501 GridInsightsV1LinesGetLatestIcingResponse200, ) - from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_apparent_power_response_200 import ( # noqa: E501 - GridInsightsV1LinesGetApparentPowerResponse200, - ) - from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_response_200 import ( # noqa: E501 - GridInsightsV1LinesGetIcingResponse200, - ) - from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_forecast_response_200 import ( # noqa: E501 - GridInsightsV1LinesGetIcingForecastResponse200, - ) - from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_latest_apparent_power_response_200 import ( # noqa: E501 - GridInsightsV1LinesGetLatestApparentPowerResponse200, - ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_sag_and_clearance_response_200 import ( # noqa: E501 GridInsightsV1LinesGetSagAndClearanceResponse200, ) diff --git a/python/heimdall_api_client/grid_insights.py b/python/heimdall_api_client/grid_insights.py index 925eeea..2ceae2f 100644 --- a/python/heimdall_api_client/grid_insights.py +++ b/python/heimdall_api_client/grid_insights.py @@ -16,9 +16,6 @@ from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_conductor_temperatures_response_200 import ( # noqa: E501 GridInsightsV1LinesGetConductorTemperaturesResponse200, ) - from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_currents_response_200 import ( # noqa: E501 - GridInsightsV1LinesGetCurrentsResponse200, - ) from heimdall_api_client.grid_insights_api_client.models.grid_insights_v1_lines_get_icing_forecast_response_200 import ( # noqa: E501 GridInsightsV1LinesGetIcingForecastResponse200, ) diff --git a/python/tests/integration/test_when_fetching_capacity_monitoring.py b/python/tests/integration/test_when_fetching_capacity_monitoring.py index e71c45a..3157d3b 100644 --- a/python/tests/integration/test_when_fetching_capacity_monitoring.py +++ b/python/tests/integration/test_when_fetching_capacity_monitoring.py @@ -11,8 +11,8 @@ _HEIMDALL_POWER_LINE_ID = uuid.UUID("d67d2205-6629-4bbd-aa9f-436bf22842ad") # "Heimdall Power Line" facility – c0ad547d-0d06-4f4c-b5dc-d319430902d2 _HEIMDALL_POWER_FACILITY_ID = uuid.UUID("c0ad547d-0d06-4f4c-b5dc-d319430902d2") -_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) -_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc) +_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC) +_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.UTC) @pytest.mark.integration diff --git a/python/tests/integration/test_when_fetching_grid_insights.py b/python/tests/integration/test_when_fetching_grid_insights.py index 64ea73b..51e88c0 100644 --- a/python/tests/integration/test_when_fetching_grid_insights.py +++ b/python/tests/integration/test_when_fetching_grid_insights.py @@ -9,8 +9,8 @@ # "Heimdall Power Line" – d67d2205-6629-4bbd-aa9f-436bf22842ad _HEIMDALL_POWER_LINE_ID = uuid.UUID("d67d2205-6629-4bbd-aa9f-436bf22842ad") -_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) -_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc) +_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC) +_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.UTC) @pytest.mark.integration @@ -217,91 +217,3 @@ def test_get_apparent_power_all_readings_should_have_timestamps_within_requested assert ap.timestamp >= _FROM, f"Timestamp {ap.timestamp} is before {_FROM}" assert ap.timestamp <= _TO, f"Timestamp {ap.timestamp} is after {_TO}" - - -# --------------------------------------------------------------------------- -# get_currents – historical -# --------------------------------------------------------------------------- - - -@pytest.fixture(scope="module") -def currents_result(api_client): - return api_client.get_currents( - line_id=_HEIMDALL_POWER_LINE_ID, - from_timestamp=_FROM, - to_timestamp=_TO, - ) - - -@pytest.mark.integration -def test_get_currents_should_return_response(currents_result): - assert currents_result is not None - - -@pytest.mark.integration -def test_get_currents_result_should_have_metric(currents_result): - assert currents_result.data.metric, "Metric should not be empty" - - -@pytest.mark.integration -def test_get_currents_result_should_have_unit(currents_result): - assert currents_result.data.unit, "Unit should not be empty" - - -@pytest.mark.integration -def test_get_currents_result_should_have_currents_list(currents_result): - # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. - assert currents_result.data.currents is not None - - -@pytest.mark.integration -def test_get_currents_all_currents_should_have_timestamps_within_requested_range(currents_result): - for current in currents_result.data.currents: - assert current.timestamp >= _FROM, f"Timestamp {current.timestamp} is before {_FROM}" - assert current.timestamp <= _TO, f"Timestamp {current.timestamp} is after {_TO}" - - -# --------------------------------------------------------------------------- -# get_conductor_temperatures – historical -# --------------------------------------------------------------------------- - - -@pytest.fixture(scope="module") -def conductor_temperatures_result(api_client): - return api_client.get_conductor_temperatures( - line_id=_HEIMDALL_POWER_LINE_ID, - from_timestamp=_FROM, - to_timestamp=_TO, - ) - - -@pytest.mark.integration -def test_get_conductor_temperatures_should_return_response(conductor_temperatures_result): - assert conductor_temperatures_result is not None - - -@pytest.mark.integration -def test_get_conductor_temperatures_result_should_have_metric(conductor_temperatures_result): - assert conductor_temperatures_result.data.metric, "Metric should not be empty" - - -@pytest.mark.integration -def test_get_conductor_temperatures_result_should_have_unit(conductor_temperatures_result): - assert conductor_temperatures_result.data.unit, "Unit should not be empty" - - -@pytest.mark.integration -def test_get_conductor_temperatures_result_should_have_list(conductor_temperatures_result): - # The API returns HTTP 200 with a (possibly empty) list – an empty list is valid. - assert conductor_temperatures_result.data.conductor_temperatures is not None - - -@pytest.mark.integration -def test_get_conductor_temperatures_all_readings_should_have_timestamps_within_requested_range( - conductor_temperatures_result, -): - for ct in conductor_temperatures_result.data.conductor_temperatures: - assert ct.timestamp >= _FROM, f"Timestamp {ct.timestamp} is before {_FROM}" - assert ct.timestamp <= _TO, f"Timestamp {ct.timestamp} is after {_TO}" - - From f945e0af0afb7b6b268b3f2eda7c0949e214f58c Mon Sep 17 00:00:00 2001 From: Joakim Amundsen Date: Thu, 2 Jul 2026 10:16:56 +0200 Subject: [PATCH 3/4] remove duplicate tests, fix build --- pr-description.md | 0 python/tests/integration/test_when_fetching_grid_insights.py | 4 ++-- 2 files changed, 2 insertions(+), 2 deletions(-) delete mode 100644 pr-description.md diff --git a/pr-description.md b/pr-description.md deleted file mode 100644 index e69de29..0000000 diff --git a/python/tests/integration/test_when_fetching_grid_insights.py b/python/tests/integration/test_when_fetching_grid_insights.py index 51e88c0..b4e8f0d 100644 --- a/python/tests/integration/test_when_fetching_grid_insights.py +++ b/python/tests/integration/test_when_fetching_grid_insights.py @@ -9,8 +9,8 @@ # "Heimdall Power Line" – d67d2205-6629-4bbd-aa9f-436bf22842ad _HEIMDALL_POWER_LINE_ID = uuid.UUID("d67d2205-6629-4bbd-aa9f-436bf22842ad") -_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC) -_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.UTC) +_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) +_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc) @pytest.mark.integration From fae1c213ab1992c70bd69b17907b34bb2be17d8c Mon Sep 17 00:00:00 2001 From: Joakim Amundsen Date: Thu, 2 Jul 2026 10:21:32 +0200 Subject: [PATCH 4/4] fix: apply ruff format and use datetime.UTC alias in integration tests --- .../capacity_monitoring.py | 7 +--- python/heimdall_api_client/grid_insights.py | 9 ++--- python/tests/integration/conftest.py | 2 - .../test_when_fetching_capacity_monitoring.py | 1 - .../test_when_fetching_grid_insights.py | 5 +-- python/tests/unit/test_latest_endpoints.py | 37 +++++-------------- 6 files changed, 16 insertions(+), 45 deletions(-) diff --git a/python/heimdall_api_client/capacity_monitoring.py b/python/heimdall_api_client/capacity_monitoring.py index 33b067b..aed7caa 100644 --- a/python/heimdall_api_client/capacity_monitoring.py +++ b/python/heimdall_api_client/capacity_monitoring.py @@ -164,8 +164,7 @@ def get_heimdall_dlrs( if response.status_code != 200: status = int(response.status_code) raise HeimdallApiError( - f"Error fetching Heimdall DLRs: {status} {response.status_code.phrase}" - f" - {body_preview(response.content)}", + f"Error fetching Heimdall DLRs: {status} {response.status_code.phrase} - {body_preview(response.content)}", status_code=status, ) return response.parsed @@ -192,8 +191,7 @@ def get_heimdall_aars( if response.status_code != 200: status = int(response.status_code) raise HeimdallApiError( - f"Error fetching Heimdall AARs: {status} {response.status_code.phrase}" - f" - {body_preview(response.content)}", + f"Error fetching Heimdall AARs: {status} {response.status_code.phrase} - {body_preview(response.content)}", status_code=status, ) return response.parsed @@ -225,4 +223,3 @@ def get_circuit_ratings( status_code=status, ) return response.parsed - diff --git a/python/heimdall_api_client/grid_insights.py b/python/heimdall_api_client/grid_insights.py index 2ceae2f..4d340f7 100644 --- a/python/heimdall_api_client/grid_insights.py +++ b/python/heimdall_api_client/grid_insights.py @@ -168,8 +168,7 @@ def get_icing( if response.status_code != 200: status = int(response.status_code) raise HeimdallApiError( - f"Error fetching icing: {status} {response.status_code.phrase}" - f" - {body_preview(response.content)}", + f"Error fetching icing: {status} {response.status_code.phrase} - {body_preview(response.content)}", status_code=status, ) return response.parsed @@ -230,8 +229,7 @@ def get_apparent_power( if response.status_code != 200: status = int(response.status_code) raise HeimdallApiError( - f"Error fetching apparent power: {status} {response.status_code.phrase}" - f" - {body_preview(response.content)}", + f"Error fetching apparent power: {status} {response.status_code.phrase} - {body_preview(response.content)}", status_code=status, ) return response.parsed @@ -278,8 +276,7 @@ def get_icing_forecast( if response.status_code != 200: status = int(response.status_code) raise HeimdallApiError( - f"Error fetching icing forecast: {status} {response.status_code.phrase}" - f" - {body_preview(response.content)}", + f"Error fetching icing forecast: {status} {response.status_code.phrase} - {body_preview(response.content)}", status_code=status, ) return response.parsed diff --git a/python/tests/integration/conftest.py b/python/tests/integration/conftest.py index bc629df..a14b773 100644 --- a/python/tests/integration/conftest.py +++ b/python/tests/integration/conftest.py @@ -10,5 +10,3 @@ def api_client(): return HeimdallApiClient( client_id=os.environ["HEIMDALL_CLIENT_ID"], client_secret=os.environ["HEIMDALL_CLIENT_SECRET"] ) - - diff --git a/python/tests/integration/test_when_fetching_capacity_monitoring.py b/python/tests/integration/test_when_fetching_capacity_monitoring.py index 3157d3b..9a0ee24 100644 --- a/python/tests/integration/test_when_fetching_capacity_monitoring.py +++ b/python/tests/integration/test_when_fetching_capacity_monitoring.py @@ -166,4 +166,3 @@ def test_get_circuit_ratings_all_circuit_ratings_should_have_timestamps_within_r def test_get_circuit_ratings_all_circuit_ratings_should_have_positive_values(circuit_ratings_result): for cr in circuit_ratings_result.data.circuit_ratings: assert cr.value > 0, f"Circuit rating value {cr.value} at {cr.timestamp} should be positive" - diff --git a/python/tests/integration/test_when_fetching_grid_insights.py b/python/tests/integration/test_when_fetching_grid_insights.py index b4e8f0d..0b5b42b 100644 --- a/python/tests/integration/test_when_fetching_grid_insights.py +++ b/python/tests/integration/test_when_fetching_grid_insights.py @@ -9,8 +9,8 @@ # "Heimdall Power Line" – d67d2205-6629-4bbd-aa9f-436bf22842ad _HEIMDALL_POWER_LINE_ID = uuid.UUID("d67d2205-6629-4bbd-aa9f-436bf22842ad") -_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) -_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc) +_FROM = datetime.datetime(2026, 1, 1, 0, 0, 0, tzinfo=datetime.UTC) +_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.UTC) @pytest.mark.integration @@ -216,4 +216,3 @@ def test_get_apparent_power_all_readings_should_have_timestamps_within_requested for ap in apparent_power_result.data.apparent_powers: assert ap.timestamp >= _FROM, f"Timestamp {ap.timestamp} is before {_FROM}" assert ap.timestamp <= _TO, f"Timestamp {ap.timestamp} is after {_TO}" - diff --git a/python/tests/unit/test_latest_endpoints.py b/python/tests/unit/test_latest_endpoints.py index 1d3eaaa..35e1070 100644 --- a/python/tests/unit/test_latest_endpoints.py +++ b/python/tests/unit/test_latest_endpoints.py @@ -101,18 +101,14 @@ def test_returns_parsed_response_on_200(self): def test_raises_404_error_on_not_found(self): client = _make_client() - with patch( - "heimdall_api_client.grid_insights.get_latest_conductor_temperature", side_effect=_raises_404() - ): + with patch("heimdall_api_client.grid_insights.get_latest_conductor_temperature", side_effect=_raises_404()): with pytest.raises(HeimdallApiError) as exc_info: client.get_latest_conductor_temperature(line_id=_LINE_ID) assert exc_info.value.status_code == 404 def test_404_is_not_retried(self): client = _make_client() - with patch( - "heimdall_api_client.grid_insights.get_latest_conductor_temperature", side_effect=_raises_404() - ): + with patch("heimdall_api_client.grid_insights.get_latest_conductor_temperature", side_effect=_raises_404()): with patch("time.sleep") as mock_sleep: with pytest.raises(HeimdallApiError): client.get_latest_conductor_temperature(line_id=_LINE_ID) @@ -310,18 +306,14 @@ def test_returns_parsed_response_on_200(self): def test_raises_404_error_on_not_found(self): client = _make_client() - with patch( - "heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", side_effect=_raises_404() - ): + with patch("heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", side_effect=_raises_404()): with pytest.raises(HeimdallApiError) as exc_info: client.get_latest_heimdall_dlr_forecasts(line_id=_LINE_ID) assert exc_info.value.status_code == 404 def test_404_is_not_retried(self): client = _make_client() - with patch( - "heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", side_effect=_raises_404() - ): + with patch("heimdall_api_client.client.get_latest_heimdall_dlr_forecasts", side_effect=_raises_404()): with patch("time.sleep") as mock_sleep: with pytest.raises(HeimdallApiError): client.get_latest_heimdall_dlr_forecasts(line_id=_LINE_ID) @@ -343,18 +335,14 @@ def test_returns_parsed_response_on_200(self): def test_raises_404_error_on_not_found(self): client = _make_client() - with patch( - "heimdall_api_client.client.get_latest_heimdall_arr_forecasts", side_effect=_raises_404() - ): + with patch("heimdall_api_client.client.get_latest_heimdall_arr_forecasts", side_effect=_raises_404()): with pytest.raises(HeimdallApiError) as exc_info: client.get_latest_heimdall_aar_forecasts(line_id=_LINE_ID) assert exc_info.value.status_code == 404 def test_404_is_not_retried(self): client = _make_client() - with patch( - "heimdall_api_client.client.get_latest_heimdall_arr_forecasts", side_effect=_raises_404() - ): + with patch("heimdall_api_client.client.get_latest_heimdall_arr_forecasts", side_effect=_raises_404()): with patch("time.sleep") as mock_sleep: with pytest.raises(HeimdallApiError): client.get_latest_heimdall_aar_forecasts(line_id=_LINE_ID) @@ -370,26 +358,20 @@ class TestGetLatestCircuitRating: def test_returns_parsed_response_on_200(self): client = _make_client() expected = MagicMock() - with patch( - "heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", return_value=expected - ): + with patch("heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", return_value=expected): result = client.get_latest_circuit_rating(facility_id=_FACILITY_ID) assert result is expected def test_raises_404_error_on_not_found(self): client = _make_client() - with patch( - "heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", side_effect=_raises_404() - ): + with patch("heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", side_effect=_raises_404()): with pytest.raises(HeimdallApiError) as exc_info: client.get_latest_circuit_rating(facility_id=_FACILITY_ID) assert exc_info.value.status_code == 404 def test_404_is_not_retried(self): client = _make_client() - with patch( - "heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", side_effect=_raises_404() - ): + with patch("heimdall_api_client.capacity_monitoring.get_latest_circuit_ratring", side_effect=_raises_404()): with patch("time.sleep") as mock_sleep: with pytest.raises(HeimdallApiError): client.get_latest_circuit_rating(facility_id=_FACILITY_ID) @@ -432,4 +414,3 @@ def test_404_is_not_retried(self): with pytest.raises(HeimdallApiError): client.get_latest_circuit_rating_forecasts(facility_id=_FACILITY_ID) mock_sleep.assert_not_called() -