From b7d4271c7ec81a33f810d3b6ff3856572d84fe4e Mon Sep 17 00:00:00 2001 From: Muhammad Gibran Alfarizi Date: Mon, 6 Jul 2026 13:51:16 +0200 Subject: [PATCH 1/2] ENH: Add an endpoint to fetch well headers from SMDA --- src/fmu_settings_api/interfaces/smda_api.py | 11 + src/fmu_settings_api/models/smda.py | 102 +++++++++ src/fmu_settings_api/services/smda.py | 71 ++++++ src/fmu_settings_api/v1/routes/smda/main.py | 61 +++++ tests/test_interfaces/test_smda_api.py | 17 ++ tests/test_services/test_smda_service.py | 126 ++++++++++- tests/test_v1/test_smda.py | 236 +++++++++++++++++++- 7 files changed, 622 insertions(+), 2 deletions(-) diff --git a/src/fmu_settings_api/interfaces/smda_api.py b/src/fmu_settings_api/interfaces/smda_api.py index 5e42d76d..6a239ee5 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 33c7698e..aae86597 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 f02a5e0c..14f89d53 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 2447d00c..2804ea53 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 5334c019..01a4b172 100644 --- a/tests/test_interfaces/test_smda_api.py +++ b/tests/test_interfaces/test_smda_api.py @@ -296,3 +296,20 @@ 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 correct payload.""" + 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 b74d9c7b..4e1777e2 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 05539bf7..9085f1ac 100644 --- a/tests/test_v1/test_smda.py +++ b/tests/test_v1/test_smda.py @@ -15,9 +15,13 @@ 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.config import HttpHeader +from fmu_settings_api.interfaces.smda_api import SmdaRoutes from fmu_settings_api.models.smda import ( SmdaFieldSearchResult, SmdaFieldUUID, @@ -1258,3 +1262,233 @@ 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, + mock_SmdaAPI_post: AsyncMock, +) -> None: + """Tests successful post to well_headers with valid field identifier.""" + well_uuid = uuid4() + wellbore_uuid = uuid4() + mock_response = MagicMock(spec=httpx2.Response) + mock_response.status_code = 200 + mock_response.json.return_value = { + "data": { + "results": [ + { + "unique_well_identifier": "NO 30/9-B", + "unique_wellbore_identifier": "NO 30/9-B-43 A", + "official_wellbore_name": "30/9-B-43 A", + "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": str(well_uuid), + "wellbore_uuid": str(wellbore_uuid), + } + ] + } + } + + mock_SmdaAPI_post.return_value = mock_response + + 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 + ) + response_data = response.json() + assert len(response_data["well_headers"]) == 1 + assert response_data["well_headers"][0]["unique_well_identifier"] == "NO 30/9-B" + assert ( + response_data["well_headers"][0]["unique_wellbore_identifier"] + == "NO 30/9-B-43 A" + ) + mock_SmdaAPI_post.assert_awaited_once_with( + SmdaRoutes.WELL_HEADER_SEARCH, + json={ + "_projection": ( + "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" + ), + "field_identifier": ["TROLL"], + }, + ) + + +async def test_post_well_headers_empty_results( + client_with_smda_session: TestClient, + session_tmp_path: Path, + mock_SmdaAPI_post: AsyncMock, +) -> None: + """Tests error when no well headers are found for a field identifier.""" + mock_response = MagicMock(spec=httpx2.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"data": {"results": []}} + + mock_SmdaAPI_post.return_value = mock_response + + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "NONEXISTENT"}, + ) + + 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"] + + +async def test_post_well_headers_drogon_uses_drogon_data( + client_with_smda_session: TestClient, + session_tmp_path: Path, + mock_SmdaAPI_post: AsyncMock, +) -> None: + """Tests that Drogon well header requests do not call SMDA and use Drogon data.""" + response = client_with_smda_session.post( + f"{ROUTE}/well_headers", + json={"identifier": "Drogon"}, + ) + + assert response.status_code == status.HTTP_200_OK, response.json() + well_header_names = [ + header["official_wellbore_name"] for header in response.json()["well_headers"] + ] + assert well_header_names == [well["name"] for well in DROGON_RMS_WELLS] + assert response.json()["well_headers"][0]["country_identifier"] == "Norway" + assert response.json()["well_headers"][0]["parent_wellbore"] is None + assert response.json()["well_headers"][0]["wellbore_purpose"] == "production" + assert response.json()["well_headers"][0]["wellbore_status"] == "operating" + assert response.json()["well_headers"][0]["multilateral"] == 0 + mlw_header = next( + header + for header in response.json()["well_headers"] + if header["official_wellbore_name"] == "MLW_OP5_Y1" + ) + assert mlw_header["parent_wellbore"] is None + assert mlw_header["multilateral"] == 1 + assert ( + response.json()["well_headers"][0]["projected_coordinate_system"] + == "ST_WGS84_UTM37N_P32637" + ) + mock_SmdaAPI_post.assert_not_awaited() + + +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.""" + 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"] + + +async def test_post_well_headers_malformed_response( + client_with_smda_session: TestClient, + session_tmp_path: Path, + mock_SmdaAPI_post: AsyncMock, +) -> None: + """Tests error handling for malformed SMDA well header response.""" + mock_response = MagicMock(spec=httpx2.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"malformed": "response"} + + mock_SmdaAPI_post.return_value = mock_response + + 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"] + + +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.""" + with patch("fmu_settings_api.deps.smda.SmdaAPI") as mock_smda_class: + mock_smda_instance = AsyncMock() + mock_smda_instance.well_headers.side_effect = TimeoutError("Request timed out") + mock_smda_class.return_value = mock_smda_instance + + 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." + + +async def test_post_well_headers_http_error( + client_with_smda_session: TestClient, + session_tmp_path: Path, + mock_SmdaAPI_post: AsyncMock, +) -> 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 + mock_SmdaAPI_post.side_effect = httpx2.HTTPStatusError( + "401 Client Error: Access Denied", + request=mock_request, + response=mock_response, + ) + + 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"] From abcea2701c0f625aef9082f4349169eb71ee8655 Mon Sep 17 00:00:00 2001 From: Muhammad Gibran Alfarizi Date: Wed, 8 Jul 2026 14:45:02 +0200 Subject: [PATCH 2/2] TST: Improve tests --- tests/test_interfaces/test_smda_api.py | 21 +++- tests/test_v1/test_smda.py | 165 +++++++++---------------- 2 files changed, 77 insertions(+), 109 deletions(-) diff --git a/tests/test_interfaces/test_smda_api.py b/tests/test_interfaces/test_smda_api.py index 01a4b172..07c9c845 100644 --- a/tests/test_interfaces/test_smda_api.py +++ b/tests/test_interfaces/test_smda_api.py @@ -299,7 +299,26 @@ async def test_smda_discovery_search(mock_httpx_post: MagicMock) -> None: async def test_smda_well_header_search(mock_httpx_post: MagicMock) -> None: - """Tests well header search sends correct payload.""" + """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"]) diff --git a/tests/test_v1/test_smda.py b/tests/test_v1/test_smda.py index 9085f1ac..f2762009 100644 --- a/tests/test_v1/test_smda.py +++ b/tests/test_v1/test_smda.py @@ -15,16 +15,15 @@ DiscoveryItem, StratigraphicColumn, ) -from fmu.settings._drogon import ( - MASTERDATA as DROGON_MASTERDATA, - RMS_WELLS as DROGON_RMS_WELLS, -) +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.interfaces.smda_api import SmdaRoutes +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" @@ -1267,40 +1266,13 @@ async def test_post_strat_units_http_error( async def test_post_well_headers_success( client_with_smda_session: TestClient, session_tmp_path: Path, - mock_SmdaAPI_post: AsyncMock, ) -> None: """Tests successful post to well_headers with valid field identifier.""" - well_uuid = uuid4() - wellbore_uuid = uuid4() - mock_response = MagicMock(spec=httpx2.Response) - mock_response.status_code = 200 - mock_response.json.return_value = { - "data": { - "results": [ - { - "unique_well_identifier": "NO 30/9-B", - "unique_wellbore_identifier": "NO 30/9-B-43 A", - "official_wellbore_name": "30/9-B-43 A", - "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": str(well_uuid), - "wellbore_uuid": str(wellbore_uuid), - } - ] - } - } - - mock_SmdaAPI_post.return_value = mock_response + 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", @@ -1312,44 +1284,24 @@ async def test_post_well_headers_success( response.headers[HttpHeader.UPSTREAM_SOURCE_KEY] == HttpHeader.UPSTREAM_SOURCE_SMDA ) - response_data = response.json() - assert len(response_data["well_headers"]) == 1 - assert response_data["well_headers"][0]["unique_well_identifier"] == "NO 30/9-B" - assert ( - response_data["well_headers"][0]["unique_wellbore_identifier"] - == "NO 30/9-B-43 A" - ) - mock_SmdaAPI_post.assert_awaited_once_with( - SmdaRoutes.WELL_HEADER_SEARCH, - json={ - "_projection": ( - "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" - ), - "field_identifier": ["TROLL"], - }, - ) + 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, - mock_SmdaAPI_post: AsyncMock, ) -> None: """Tests error when no well headers are found for a field identifier.""" - mock_response = MagicMock(spec=httpx2.Response) - mock_response.status_code = 200 - mock_response.json.return_value = {"data": {"results": []}} - - mock_SmdaAPI_post.return_value = mock_response + 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": "NONEXISTENT"}, + json={"identifier": "TROLL"}, ) assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT, ( @@ -1360,41 +1312,27 @@ async def test_post_well_headers_empty_results( == 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_uses_drogon_data( +async def test_post_well_headers_drogon_identifier_calls_service( client_with_smda_session: TestClient, session_tmp_path: Path, - mock_SmdaAPI_post: AsyncMock, ) -> None: - """Tests that Drogon well header requests do not call SMDA and use Drogon data.""" + """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() - well_header_names = [ - header["official_wellbore_name"] for header in response.json()["well_headers"] - ] - assert well_header_names == [well["name"] for well in DROGON_RMS_WELLS] - assert response.json()["well_headers"][0]["country_identifier"] == "Norway" - assert response.json()["well_headers"][0]["parent_wellbore"] is None - assert response.json()["well_headers"][0]["wellbore_purpose"] == "production" - assert response.json()["well_headers"][0]["wellbore_status"] == "operating" - assert response.json()["well_headers"][0]["multilateral"] == 0 - mlw_header = next( - header - for header in response.json()["well_headers"] - if header["official_wellbore_name"] == "MLW_OP5_Y1" - ) - assert mlw_header["parent_wellbore"] is None - assert mlw_header["multilateral"] == 1 - assert ( - response.json()["well_headers"][0]["projected_coordinate_system"] - == "ST_WGS84_UTM37N_P32637" - ) - mock_SmdaAPI_post.assert_not_awaited() + smda_service.get_well_headers.assert_awaited_once_with("Drogon") async def test_post_well_headers_empty_identifier( @@ -1402,6 +1340,12 @@ async def test_post_well_headers_empty_identifier( 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": ""}, @@ -1413,19 +1357,17 @@ async def test_post_well_headers_empty_identifier( == 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, - mock_SmdaAPI_post: AsyncMock, ) -> None: """Tests error handling for malformed SMDA well header response.""" - mock_response = MagicMock(spec=httpx2.Response) - mock_response.status_code = 200 - mock_response.json.return_value = {"malformed": "response"} - - mock_SmdaAPI_post.return_value = mock_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", @@ -1440,6 +1382,7 @@ async def test_post_well_headers_malformed_response( == 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( @@ -1447,15 +1390,16 @@ async def test_post_well_headers_request_timeout( session_tmp_path: Path, ) -> None: """Tests when a well header request to SMDA times out.""" - with patch("fmu_settings_api.deps.smda.SmdaAPI") as mock_smda_class: - mock_smda_instance = AsyncMock() - mock_smda_instance.well_headers.side_effect = TimeoutError("Request timed out") - mock_smda_class.return_value = mock_smda_instance + 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"}, - ) + 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 ( @@ -1463,23 +1407,27 @@ async def test_post_well_headers_request_timeout( == 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, - mock_SmdaAPI_post: AsyncMock, ) -> 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 - mock_SmdaAPI_post.side_effect = httpx2.HTTPStatusError( - "401 Client Error: Access Denied", - request=mock_request, - response=mock_response, + 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", @@ -1492,3 +1440,4 @@ async def test_post_well_headers_http_error( == HttpHeader.UPSTREAM_SOURCE_SMDA ) assert "SMDA error requesting" in response.json()["detail"] + smda_service.get_well_headers.assert_awaited_once_with("TROLL")