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..aed7caa 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,85 @@ 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} - {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} - {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..03f0b71 100644
--- a/python/heimdall_api_client/client.py
+++ b/python/heimdall_api_client/client.py
@@ -14,6 +14,9 @@
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,
@@ -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,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,
)
@@ -55,6 +85,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,
+ )
_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..4d340f7 100644
--- a/python/heimdall_api_client/grid_insights.py
+++ b/python/heimdall_api_client/grid_insights.py
@@ -10,6 +10,21 @@
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_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 +34,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 +139,172 @@ 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} - {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} - {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} - {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/test_when_fetching_capacity_monitoring.py b/python/tests/integration/test_when_fetching_capacity_monitoring.py
new file mode 100644
index 0000000..9a0ee24
--- /dev/null
+++ b/python/tests/integration/test_when_fetching_capacity_monitoring.py
@@ -0,0 +1,168 @@
+"""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.UTC)
+_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.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..0b5b42b
--- /dev/null
+++ b/python/tests/integration/test_when_fetching_grid_insights.py
@@ -0,0 +1,218 @@
+"""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.UTC)
+_TO = datetime.datetime(2026, 1, 2, 0, 0, 0, tzinfo=datetime.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}"
diff --git a/python/tests/unit/test_latest_endpoints.py b/python/tests/unit/test_latest_endpoints.py
new file mode 100644
index 0000000..35e1070
--- /dev/null
+++ b/python/tests/unit/test_latest_endpoints.py
@@ -0,0 +1,416 @@
+"""
+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()