Commit 707c2557 authored by Gregory Harris's avatar Gregory Harris
Browse files

Adding some unit tests

parent dc105b2b
......@@ -7,6 +7,7 @@ from lasloader.record_mapper import Record
class TestOsduClient:
_expected_ids = [123, 234, 345]
_expected_ids_str = [{"id": "321"}, {"id": "432"}, {"id": "543"}]
_access_token = "token"
_data_partition_id = "AnId"
......@@ -20,7 +21,7 @@ class TestOsduClient:
if "data-partition-id" not in request.headers or self._data_partition_id != request.headers["data-partition-id"]:
return httpx.Response(500, text="No data-partition-id header")
return httpx.Response(200, json={"recordIds": self._expected_ids})
return httpx.Response(200, json={"recordIds": self._expected_ids, "totalCount": 3, "results": self._expected_ids_str})
def match_and_mock_get(self, request: any):
if request.method != "GET":
......@@ -290,3 +291,18 @@ class TestOsduClient:
result = client.post_log_recognition("mnemonic", "unit")
assert result is not None
def test_search_for_wellbore(self, respx_mock):
# Assemble
base_url = "http://test.bp.com"
url = f"{base_url}/api/search/v2/query"
respx_mock.post(url).mock(side_effect=self.match_and_mock_post)
client = OsduClient(base_url, self._access_token, self._data_partition_id)
# Act
result = client.search_for_wellbore("Wellbore Name")
assert result == ["321", "432", "543"]
from pandas.core.frame import DataFrame
import pytest
from unittest.mock import Mock, ANY
from lasloader.well_service import WellBoreService, WellLogService
from lasloader.well_service import WellBoreService, WellLogService, LasLoaderConflictError
from lasloader.osdu_client import OsduClient, LasLoaderWebResponseError
from lasloader.record_mapper import LasToRecordMapper, Record, WellLogRecord, WellboreRecord
......@@ -266,14 +266,14 @@ class TestWellLogService:
class TestWellBoreService:
@pytest.mark.parametrize("no_recognize", [True, False])
def test_recognize_log_family(self, no_recognize: bool):
@pytest.mark.parametrize("no_recognize,existing_wellbore_ids", [(True, None), (False, None), (True, ["well_bore_id"])])
def test_file_ingest(self, no_recognize: bool, existing_wellbore_ids: list[str]):
# Assemble
mock_client = Mock(spec=OsduClient)
mock_well_log_service = Mock(spec=WellLogService)
mock_mapper = Mock(spec=LasToRecordMapper)
well_bore_record = WellboreRecord({"Kind": "well_bore_kind", "acl": {}, "legal": {}, "data": {}})
well_bore_record = WellboreRecord({"Kind": "well_bore_kind", "acl": {}, "legal": {}, "data": {"FacilityName": "WellBoreName"}})
well_log_record = Record("well_log_kind", {}, {}, {})
well_log_record_recognized = Record("well_log_rec_kind", {"blah": "blah"}, {}, {})
well_log_record_returned = WellLogRecord({"kind": "returned_welllog_record"})
......@@ -287,6 +287,7 @@ class TestWellBoreService:
mock_mapper.map_to_well_log_record.return_value = well_log_record
mock_mapper.extract_log_data.return_value = well_log_data
mock_client.search_for_wellbore.return_value = existing_wellbore_ids
mock_client.create_wellbore.return_value = well_bore_ids
mock_client.get_wellbore_record.return_value = well_bore_record
mock_client.post_welllog.return_value = well_log_ids
......@@ -300,18 +301,91 @@ class TestWellBoreService:
subject.file_ingest(mock_mapper, data_partition_id, no_recognize)
# Assert
mock_mapper.map_to_wellbore_record.called_once()
mock_client.create_wellbore.called_once_with(well_bore_record)
mock_client.get_wellbore_record.called_once_with(well_bore_ids[0])
mock_mapper.map_to_well_log_record.called_once_with(well_bore_ids[0])
mock_client.search_for_wellbore.assert_called_once_with("WellBoreName")
mock_mapper.map_to_wellbore_record.assert_called_once()
if existing_wellbore_ids is None:
mock_client.create_wellbore.assert_called_once_with(well_bore_record)
else:
mock_client.create_wellbore.assert_not_called()
mock_client.get_wellbore_record.assert_called_once_with(well_bore_ids[0])
mock_mapper.map_to_well_log_record.assert_called_once_with(well_bore_ids[0])
if no_recognize:
mock_well_log_service.recognize_log_family.assert_not_called()
mock_client.post_welllog.called_once_with(well_log_record)
assert not mock_well_log_service.recognize_log_family.called
mock_client.create_welllog.assert_called_once_with(well_log_record)
else:
mock_well_log_service.recognize_log_family.called_once_with(well_log_record, data_partition_id)
mock_client.post_welllog.called_once_with(well_log_record_recognized)
mock_well_log_service.recognize_log_family.assert_called_once_with(well_log_record, data_partition_id)
mock_client.create_welllog.assert_called_once_with(well_log_record_recognized)
mock_client.get_welllog_record.assert_called_once_with(well_log_ids[0])
mock_mapper.extract_log_data.assert_called_once()
mock_client.add_welllog_data.assert_called_once_with(well_log_data, well_log_ids[0])
def test_file_ingest_raises_on_well_bore_conflict(self):
# Assemble
mock_client = Mock(spec=OsduClient)
mock_well_log_service = Mock(spec=WellLogService)
mock_mapper = Mock(spec=LasToRecordMapper)
mock_client.get_welllog_record.called_once_with(well_log_ids[0])
mock_mapper.extract_log_data.called_once()
mock_client.add_welllog_data.called_once_with(well_log_data, well_log_ids[0])
well_bore_record = WellboreRecord({"Kind": "well_bore_kind", "acl": {}, "legal": {}, "data": {"FacilityName": "WellBoreName"}})
data_partition_id = "dp_id"
mock_mapper.map_to_wellbore_record.return_value = well_bore_record
mock_client.search_for_wellbore.return_value = ["id1", "id2"]
subject = WellBoreService(mock_client, mock_well_log_service)
# Act
with pytest.raises(LasLoaderConflictError):
subject.file_ingest(mock_mapper, data_partition_id, False)
# Assert
mock_client.search_for_wellbore.assert_called_once_with("WellBoreName")
mock_mapper.map_to_wellbore_record.assert_called_once()
mock_client.create_wellbore.assert_not_called()
mock_client.create_welllog.assert_not_called()
mock_client.add_welllog_data.assert_not_called()
def test_get_wellbore_by_name_raises_on_conflict(self):
# Assemble
mock_client = Mock(spec=OsduClient)
mock_well_log_service = Mock(spec=WellLogService)
mock_client.search_for_wellbore.return_value = ["id1", "id2"]
subject = WellBoreService(mock_client, mock_well_log_service)
# Act
with pytest.raises(LasLoaderConflictError):
subject._get_wellbore_by_name("WellBoreName")
# Assert
mock_client.search_for_wellbore.assert_called_once_with("WellBoreName")
@pytest.mark.parametrize("existing_wellbore_ids,wellbore_name",
[(None, "WellBoreName"), (["well_bore_id"], "WellBoreName"), (None, None)])
def test_get_wellbore_by_name(self, existing_wellbore_ids, wellbore_name):
# Assemble
mock_client = Mock(spec=OsduClient)
mock_well_log_service = Mock(spec=WellLogService)
mock_client.search_for_wellbore.return_value = existing_wellbore_ids
subject = WellBoreService(mock_client, mock_well_log_service)
# Act
result = subject._get_wellbore_by_name(wellbore_name)
# Assert
if wellbore_name is None:
mock_client.search_for_wellbore.assert_not_called()
else:
mock_client.search_for_wellbore.assert_called_once_with(wellbore_name)
if existing_wellbore_ids is None or wellbore_name is None:
assert result is None
else:
assert result == existing_wellbore_ids[0]
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment