diff --git a/src/fmu_settings_api/interfaces/smda_api.py b/src/fmu_settings_api/interfaces/smda_api.py index 5e42d76..6a239ee 100644 --- a/src/fmu_settings_api/interfaces/smda_api.py +++ b/src/fmu_settings_api/interfaces/smda_api.py @@ -21,6 +21,7 @@ class SmdaRoutes: STRAT_UNITS_SEARCH: Final[str] = "smda-api/strat-units/search" COORDINATE_SYSTEM_SEARCH: Final[str] = "smda-api/crs/search" SURFACE_SEARCH: Final[str] = "smda-api/strat-surface-names/search" + WELL_HEADER_SEARCH: Final[str] = "smda-api/wellheader-insight-headers/search" class SmdaAPI: @@ -113,6 +114,16 @@ async def discovery( json={"_projection": _projection, "field_identifier": field_identifiers}, ) + async def well_headers( + self, field_identifiers: Sequence[str], columns: Sequence[str] | None = None + ) -> httpx2.Response: + """Searches for well headers related to a field identifier.""" + _projection = "identifier,uuid" if columns is None else ",".join(columns) + return await self.post( + SmdaRoutes.WELL_HEADER_SEARCH, + json={"_projection": _projection, "field_identifier": field_identifiers}, + ) + async def strat_column_areas( self, field_identifiers: Sequence[str], columns: Sequence[str] | None = None ) -> httpx2.Response: diff --git a/src/fmu_settings_api/models/smda.py b/src/fmu_settings_api/models/smda.py index 33c7698..aae8659 100644 --- a/src/fmu_settings_api/models/smda.py +++ b/src/fmu_settings_api/models/smda.py @@ -63,6 +63,101 @@ class SmdaFieldSearchResult(BaseResponseModel): """A list of field identifier results from the search.""" +class SmdaWellHeader(BaseResponseModel): + """Well header data from SMDA.""" + + unique_well_identifier: str + """Unique SMDA identifier for the well.""" + + unique_wellbore_identifier: str + """Unique SMDA identifier for the wellbore.""" + + official_wellbore_name: str | None + """Official wellbore name used by the Authorities. + + For Norway and UK, it will be the unique_wellbore_identifier without + country iso code, but for Brazil it can really differs from the Equinor + wellbore name. + """ + + country_identifier: str + """Country identifier for the wellbore.""" + + parent_wellbore: str | None + """The unique wellbore identifier this wellbore is kicked off from. + + Ref. kick off depth. This is used for sidetracks. A wellbore starting at + the well origin has no parent. + """ + + wellbore_type: str | None + """Type of wellbore, values like exploration, development, other. + + This attribute is automatically maintained in SMDA based on the wellbore + purpose. If the purpose is like wildcat or appraisal, type will be set to + exploration, if the purpose is like production, injection then the type is + set to development. + """ + + wellbore_purpose: str | None + """Purpose of wellbore. + + Values like wildcat, appraisal, … for exploration wellbores; production, + injection, observation, disposal, … for development wellbores; shallow gas, + pilot hole for other purpose. + """ + + wellbore_status: str | None + """Status of the wellbore. + + Value like plugged and abandoned, drilling, plugged, producing ... This + attribute is automatically maintained in SMDA if no good source is found + for it. SMDA will use the wellbore type (exploration or development), the + drill dates information, current_track, etc ... in order to set a plausible + status. If wellbore type=exploration and completed_date < current_date, + then status=plugged and abandoned while development wellbore would be set + to completed. + """ + + wellbore_purpose_planned: str | None + """Pre-drill purpose of the wellbore. + + Legal values for exploration wellbores: wildcat, appraisal. Example of + legal values for development wellbores: observation, production, injection. + """ + + drill_year: int | None + """The year when the drilling has started.""" + + completion_date: str | None + """Date when the wellbore is considered completed. + + For exploration wellbores from moveable facilities, this may be the anchor + handling or jacking-down start date. For fixed facilities and development + wellbores, it is when the wellbore reaches total depth and the last casing, + liner, or screen is set. If immediately plugged, it is the date the last + plug is set. + """ + + discovery_internal_identifier: str | None + """Internal name of the discovery.""" + + multilateral: Literal[0, 1] | None + """Whether the wellbore is multilateral. 0 = no, 1 = yes.""" + + projected_coordinate_unit: str | None + """Projected coordinate unit.""" + + projected_coordinate_system: str | None + """Coordinate reference system for the easting/northing values.""" + + well_uuid: UUID + """SMDA UUID for the well.""" + + wellbore_uuid: UUID + """SMDA UUID for the wellbore.""" + + class SmdaMasterdataResult(BaseResponseModel): """Contains SMDA-related attributes.""" @@ -152,3 +247,10 @@ class SmdaStratigraphicUnitsResult(BaseResponseModel): stratigraphic_units: list[StratigraphicUnit] """List of stratigraphic units from SMDA.""" + + +class SmdaWellHeadersResult(BaseResponseModel): + """Result containing a list of well headers.""" + + well_headers: list[SmdaWellHeader] + """List of well headers from SMDA.""" diff --git a/src/fmu_settings_api/services/smda.py b/src/fmu_settings_api/services/smda.py index f02a5e0..14f89d5 100644 --- a/src/fmu_settings_api/services/smda.py +++ b/src/fmu_settings_api/services/smda.py @@ -3,6 +3,7 @@ import asyncio from collections.abc import Sequence from typing import Any, Final +from uuid import NAMESPACE_URL, uuid5 import httpx2 from fmu.datamodels.common.masterdata import ( @@ -14,6 +15,7 @@ ) from fmu.settings._drogon import ( MASTERDATA as DROGON_MASTERDATA, + RMS_WELLS as DROGON_RMS_WELLS, RMS_ZONES as DROGON_RMS_ZONES, STRATIGRAPHY_MAPPINGS as DROGON_STRATIGRAPHY_MAPPINGS, ) @@ -27,6 +29,8 @@ SmdaMasterdataResult, SmdaSelectedField, SmdaStratigraphicUnitsResult, + SmdaWellHeader, + SmdaWellHeadersResult, StratigraphicUnit, ) @@ -68,6 +72,31 @@ for zone in DROGON_RMS_ZONES if zone["name"] in DROGON_STRATIGRAPHY_BY_SOURCE_ID ] +DROGON_WELL_HEADERS: Final[list[SmdaWellHeader]] = [ + SmdaWellHeader( + unique_well_identifier=f"NO {well['name']}", + unique_wellbore_identifier=f"NO {well['name']}", + official_wellbore_name=str(well["name"]), + country_identifier=DROGON_SMDA_MASTERDATA["country"][0]["identifier"], + parent_wellbore=None, + wellbore_type="development", + wellbore_purpose="production", + wellbore_status="operating", + wellbore_purpose_planned="production", + drill_year=None, + completion_date=None, + discovery_internal_identifier=None, + multilateral=1 if well["name"] == "MLW_OP5_Y1" else 0, + projected_coordinate_unit="m", + projected_coordinate_system=DROGON_SMDA_MASTERDATA["coordinate_system"][ + "identifier" + ], + # Synthetic but stable UUIDs for the built-in Drogon data. + well_uuid=uuid5(NAMESPACE_URL, f"well/{well['name']}"), + wellbore_uuid=uuid5(NAMESPACE_URL, f"wellbore/{well['name']}"), + ) + for well in DROGON_RMS_WELLS +] def _is_drogon_identifier(identifier: str) -> bool: @@ -342,6 +371,48 @@ async def get_stratigraphic_units( strat_unit_items.append(strat_unit_item) return SmdaStratigraphicUnitsResult(stratigraphic_units=strat_unit_items) + async def get_well_headers(self, field_identifier: str) -> SmdaWellHeadersResult: + """Queries well headers for a field identifier.""" + if not field_identifier: + raise ValueError("A field identifier must be provided") + + if _is_drogon_identifier(field_identifier): + return SmdaWellHeadersResult(well_headers=DROGON_WELL_HEADERS) + + well_header_items = [] + well_header_res = await self._smda.well_headers( + [field_identifier], + columns=[ + "unique_well_identifier", + "unique_wellbore_identifier", + "official_wellbore_name", + "country_identifier", + "parent_wellbore", + "wellbore_type", + "wellbore_purpose", + "wellbore_status", + "wellbore_purpose_planned", + "drill_year", + "completion_date", + "discovery_internal_identifier", + "multilateral", + "projected_coordinate_unit", + "projected_coordinate_system", + "well_uuid", + "wellbore_uuid", + ], + ) + well_header_results = well_header_res.json()["data"]["results"] + for well_header_data in well_header_results: + well_header_item = SmdaWellHeader(**well_header_data) + if well_header_item not in well_header_items: + well_header_items.append(well_header_item) + if not well_header_items: + raise ValueError( + f"No well headers found for field identifier: {field_identifier}" + ) + return SmdaWellHeadersResult(well_headers=well_header_items) + async def _get_countries( self, country_identifiers: Sequence[str] ) -> list[CountryItem]: diff --git a/src/fmu_settings_api/v1/routes/smda/main.py b/src/fmu_settings_api/v1/routes/smda/main.py index 2447d00..2804ea5 100644 --- a/src/fmu_settings_api/v1/routes/smda/main.py +++ b/src/fmu_settings_api/v1/routes/smda/main.py @@ -20,6 +20,7 @@ SmdaSelectedField, SmdaStratColumn, SmdaStratigraphicUnitsResult, + SmdaWellHeadersResult, ) from fmu_settings_api.v1.responses import ( GetSessionResponses, @@ -34,6 +35,7 @@ [ {"detail": "At least one SMDA field must be provided"}, {"detail": "A stratigraphic column identifier must be provided"}, + {"detail": "A field identifier must be provided"}, ], ), **inline_add_response( @@ -55,6 +57,7 @@ [ {"detail": "No fields found for identifiers: {identifiers}"}, {"detail": "No stratigraphic units found for column: {identifier}"}, + {"detail": "No well headers found for field identifier: {identifier}"}, ], ), **inline_add_response( @@ -341,3 +344,61 @@ async def post_strat_units( detail="SMDA API request timed out. Please try again.", headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA}, ) from e + + +@router.post( + "/well_headers", + response_model=SmdaWellHeadersResult, + summary="Retrieves well headers for a field", + description=dedent( + """ + A route to gather well header data from SMDA for a specified field. + + This route receives a valid field identifier and returns the well header + fields needed to identify and map SMDA wellbores against model well names. + """ + ), + responses={ + **GetSessionResponses, + **SmdaUpstreamErrorResponses, + }, +) +async def post_well_headers( + field: SmdaField, + smda_service: ProjectSmdaServiceDep, +) -> SmdaWellHeadersResult: + """Queries SMDA well headers for a specified field.""" + try: + return await smda_service.get_well_headers(field.identifier) + except ValueError as e: + error_msg = str(e) + match error_msg: + case msg if "A field identifier must be provided" in msg: + status_code = 400 + case msg if "No well headers found" in msg: + status_code = 422 + case _: # pragma: no cover + status_code = 400 + raise HTTPException( + status_code=status_code, + detail=error_msg, + headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA}, + ) from e + except httpx2.HTTPStatusError as e: + raise HTTPException( + status_code=e.response.status_code, + detail=f"SMDA error requesting {e.request.url}", + headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA}, + ) from e + except KeyError as e: + raise HTTPException( + status_code=500, + detail=f"Malformed response from SMDA: {e}", + headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA}, + ) from e + except TimeoutError as e: + raise HTTPException( + status_code=503, + detail="SMDA API request timed out. Please try again.", + headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA}, + ) from e diff --git a/tests/test_interfaces/test_smda_api.py b/tests/test_interfaces/test_smda_api.py index 5334c01..07c9c84 100644 --- a/tests/test_interfaces/test_smda_api.py +++ b/tests/test_interfaces/test_smda_api.py @@ -296,3 +296,39 @@ async def test_smda_discovery_search(mock_httpx_post: MagicMock) -> None: }, ) res.raise_for_status.assert_called_once() # type: ignore + + +async def test_smda_well_header_search(mock_httpx_post: MagicMock) -> None: + """Tests well header search sends default payload.""" + api = SmdaAPI("token", "key") + + res = await api.well_headers(["FIELD_A"]) + + mock_httpx_post.assert_called_with( + f"{SmdaRoutes.BASE_URL}/{SmdaRoutes.WELL_HEADER_SEARCH}", + headers=api._headers, + json={ + "_projection": "identifier,uuid", + "field_identifier": ["FIELD_A"], + }, + ) + res.raise_for_status.assert_called_once() # type: ignore + + +async def test_smda_well_header_search_with_columns( + mock_httpx_post: MagicMock, +) -> None: + """Tests well header search sends requested projection.""" + api = SmdaAPI("token", "key") + + res = await api.well_headers(["FIELD_A"], columns=["unique_well_identifier"]) + + mock_httpx_post.assert_called_with( + f"{SmdaRoutes.BASE_URL}/{SmdaRoutes.WELL_HEADER_SEARCH}", + headers=api._headers, + json={ + "_projection": "unique_well_identifier", + "field_identifier": ["FIELD_A"], + }, + ) + res.raise_for_status.assert_called_once() # type: ignore diff --git a/tests/test_services/test_smda_service.py b/tests/test_services/test_smda_service.py index b74d9c7..4e1777e 100644 --- a/tests/test_services/test_smda_service.py +++ b/tests/test_services/test_smda_service.py @@ -12,7 +12,10 @@ DiscoveryItem, StratigraphicColumn, ) -from fmu.settings._drogon import MASTERDATA as DROGON_MASTERDATA +from fmu.settings._drogon import ( + MASTERDATA as DROGON_MASTERDATA, + RMS_WELLS as DROGON_RMS_WELLS, +) from fmu_settings_api.models.smda import SmdaField, SmdaSelectedField from fmu_settings_api.services.smda import SmdaService @@ -64,6 +67,127 @@ async def test_get_countries(given: list[str], mock_val: list[CountryItem]) -> N assert res == mock_val +async def test_get_well_headers_result() -> None: + """Tests public well header lookup queries and deduplicates well header data.""" + mock_smda = AsyncMock() + well_header_resp = MagicMock() + well_uuid = UUID("15ce3b84-766f-4c93-9050-b154861f9100") + wellbore_uuid = UUID("25ce3b84-766f-4c93-9050-b154861f9100") + well_header_data = { + "unique_well_identifier": "NO 30/9-B", + "unique_wellbore_identifier": "NO 30/9-B-43 A", + "official_wellbore_name": None, + "country_identifier": "Norway", + "parent_wellbore": None, + "wellbore_type": "initial", + "wellbore_purpose": "production", + "wellbore_status": "completed", + "wellbore_purpose_planned": "production", + "drill_year": 2024, + "completion_date": "2024-10-01", + "discovery_internal_identifier": "DISC-1", + "multilateral": 0, + "projected_coordinate_unit": "m", + "projected_coordinate_system": "ST_WGS84_UTM37N_P32637", + "well_uuid": well_uuid, + "wellbore_uuid": wellbore_uuid, + } + well_header_resp.json.return_value = { + "data": {"results": [well_header_data, well_header_data]} + } + mock_smda.well_headers.return_value = well_header_resp + + service = SmdaService(mock_smda) + res = await service.get_well_headers("TROLL") + + mock_smda.well_headers.assert_called_with( + ["TROLL"], + columns=[ + "unique_well_identifier", + "unique_wellbore_identifier", + "official_wellbore_name", + "country_identifier", + "parent_wellbore", + "wellbore_type", + "wellbore_purpose", + "wellbore_status", + "wellbore_purpose_planned", + "drill_year", + "completion_date", + "discovery_internal_identifier", + "multilateral", + "projected_coordinate_unit", + "projected_coordinate_system", + "well_uuid", + "wellbore_uuid", + ], + ) + assert len(res.well_headers) == 1 + assert res.well_headers[0].unique_well_identifier == "NO 30/9-B" + assert res.well_headers[0].unique_wellbore_identifier == "NO 30/9-B-43 A" + assert res.well_headers[0].official_wellbore_name is None + assert res.well_headers[0].well_uuid == well_uuid + assert res.well_headers[0].wellbore_uuid == wellbore_uuid + + +async def test_get_well_headers_empty_identifier() -> None: + """Tests that well header lookup requires a field identifier.""" + mock_smda = AsyncMock() + service = SmdaService(mock_smda) + + with pytest.raises(ValueError, match="A field identifier must be provided"): + await service.get_well_headers("") + + mock_smda.well_headers.assert_not_called() + + +async def test_get_well_headers_no_results() -> None: + """Tests that well header lookup raises when no results are found.""" + mock_smda = AsyncMock() + well_header_resp = MagicMock() + well_header_resp.json.return_value = {"data": {"results": []}} + mock_smda.well_headers.return_value = well_header_resp + + service = SmdaService(mock_smda) + + with pytest.raises( + ValueError, + match="No well headers found for field identifier: TROLL", + ): + await service.get_well_headers("TROLL") + + +async def test_get_drogon_well_headers_uses_drogon_data() -> None: + """Tests that Drogon well header lookup does not call SMDA and uses Drogon data.""" + mock_smda = AsyncMock() + service = SmdaService(mock_smda) + + res = await service.get_well_headers("Drogon") + + mock_smda.well_headers.assert_not_called() + assert [header.official_wellbore_name for header in res.well_headers] == [ + well["name"] for well in DROGON_RMS_WELLS + ] + assert [header.unique_well_identifier for header in res.well_headers] == [ + f"NO {well['name']}" for well in DROGON_RMS_WELLS + ] + assert [header.unique_wellbore_identifier for header in res.well_headers] == [ + f"NO {well['name']}" for well in DROGON_RMS_WELLS + ] + assert res.well_headers[0].country_identifier == "Norway" + assert res.well_headers[0].projected_coordinate_system == "ST_WGS84_UTM37N_P32637" + assert res.well_headers[0].parent_wellbore is None + assert res.well_headers[0].wellbore_purpose == "production" + assert res.well_headers[0].wellbore_status == "operating" + assert res.well_headers[0].multilateral == 0 + multilateral_well = next( + header + for header in res.well_headers + if header.official_wellbore_name == "MLW_OP5_Y1" + ) + assert multilateral_well.multilateral == 1 + + @pytest.mark.parametrize( "given, mock_val", [ diff --git a/tests/test_v1/test_smda.py b/tests/test_v1/test_smda.py index 05539bf..f276200 100644 --- a/tests/test_v1/test_smda.py +++ b/tests/test_v1/test_smda.py @@ -17,10 +17,13 @@ ) from fmu.settings._drogon import MASTERDATA as DROGON_MASTERDATA +from fmu_settings_api.__main__ import app from fmu_settings_api.config import HttpHeader +from fmu_settings_api.deps.smda import get_project_smda_service from fmu_settings_api.models.smda import ( SmdaFieldSearchResult, SmdaFieldUUID, + SmdaWellHeadersResult, ) ROUTE = "/api/v1/smda" @@ -1258,3 +1261,183 @@ async def test_post_strat_units_http_error( == HttpHeader.UPSTREAM_SOURCE_SMDA ) assert "SMDA error requesting" in response.json()["detail"] + + +async def test_post_well_headers_success( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests successful post to well_headers with valid field identifier.""" + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock( + return_value=SmdaWellHeadersResult(well_headers=[]) + ) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "TROLL"}, + ) + + assert response.status_code == status.HTTP_200_OK, response.json() + assert ( + response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] + == HttpHeader.UPSTREAM_SOURCE_SMDA + ) + assert response.json() == {"well_headers": []} + smda_service.get_well_headers.assert_awaited_once_with("TROLL") + + +async def test_post_well_headers_empty_results( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests error when no well headers are found for a field identifier.""" + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock( + side_effect=ValueError("No well headers found for field identifier: TROLL") + ) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "TROLL"}, + ) + + assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT, ( + response.json() + ) + assert ( + response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] + == HttpHeader.UPSTREAM_SOURCE_SMDA + ) + assert "No well headers found" in response.json()["detail"] + smda_service.get_well_headers.assert_awaited_once_with("TROLL") + + +async def test_post_well_headers_drogon_identifier_calls_service( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests posting Drogon calls the well header service with Drogon.""" + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock( + return_value=SmdaWellHeadersResult(well_headers=[]) + ) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "Drogon"}, + ) + + assert response.status_code == status.HTTP_200_OK, response.json() + smda_service.get_well_headers.assert_awaited_once_with("Drogon") + + +async def test_post_well_headers_empty_identifier( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests 400 error when empty field identifier is provided.""" + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock( + side_effect=ValueError("A field identifier must be provided") + ) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": ""}, + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST, response.json() + assert ( + response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] + == HttpHeader.UPSTREAM_SOURCE_SMDA + ) + assert "must be provided" in response.json()["detail"] + smda_service.get_well_headers.assert_awaited_once_with("") + + +async def test_post_well_headers_malformed_response( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests error handling for malformed SMDA well header response.""" + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock(side_effect=KeyError("data")) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "TROLL"}, + ) + + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR, ( + response.json() + ) + assert ( + response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] + == HttpHeader.UPSTREAM_SOURCE_SMDA + ) + assert "Malformed response from SMDA" in response.json()["detail"] + smda_service.get_well_headers.assert_awaited_once_with("TROLL") + + +async def test_post_well_headers_request_timeout( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests when a well header request to SMDA times out.""" + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock( + side_effect=TimeoutError("Request timed out") + ) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "TROLL"}, + ) + + assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE, response.json() + assert ( + response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] + == HttpHeader.UPSTREAM_SOURCE_SMDA + ) + assert response.json()["detail"] == "SMDA API request timed out. Please try again." + smda_service.get_well_headers.assert_awaited_once_with("TROLL") + + +async def test_post_well_headers_http_error( + client_with_smda_session: TestClient, + session_tmp_path: Path, +) -> None: + """Tests when SMDA returns HTTP error for well headers.""" + mock_request = MagicMock(spec=httpx2.Request) + mock_request.url = "https://smda/wellheaders" + mock_response = MagicMock(spec=httpx2.Response) + mock_response.status_code = 401 + smda_service = MagicMock() + smda_service.get_well_headers = AsyncMock( + side_effect=httpx2.HTTPStatusError( + "401 Client Error: Access Denied", + request=mock_request, + response=mock_response, + ) + ) + app.dependency_overrides[get_project_smda_service] = lambda: smda_service + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "TROLL"}, + ) + + assert response.status_code == status.HTTP_401_UNAUTHORIZED, response.json() + assert ( + response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] + == HttpHeader.UPSTREAM_SOURCE_SMDA + ) + assert "SMDA error requesting" in response.json()["detail"] + smda_service.get_well_headers.assert_awaited_once_with("TROLL")