diff --git a/services/util.py b/services/util.py index cb3d8826c..36c1bf7a6 100644 --- a/services/util.py +++ b/services/util.py @@ -5,6 +5,7 @@ import httpx from constants import SRID_WGS84 +from db import Base TRANSFORMERS = {} @@ -115,6 +116,45 @@ def get_epqs_elevation_from_point(lon: float, lat: float) -> float | None: return data["value"] +def retrieve_polymorphic_table_record( + target_record: Base, + polymorphic_relationship: str, + polymorphic_type: str, + latest=True, +) -> Base: + """ + Retrieve a record from a polymorphic table. This function assumes that the + parent class has the correct mixin to support retrieval via an attribute. + + Parameters: + ---------- + target_record : Base + The parent record from which to retrieve the polymorphic child record. + + polymorphic_relationship : str + The name of the relationship attribute on the parent record that corresponds to the polymorphic table. + + polymorphic_type : str + The specific type of the polymorphic record to retrieve (e.g., 'Use Status' or 'Monitoring Status' for StatusHistory). + + latest : bool, optional + If True, retrieves the latest record based on start_date. Defaults to True. + """ + if polymorphic_relationship == "permissions": + type_field = "permission_type" + elif polymorphic_relationship == "status_history": + type_field = "status_type" + + polymorphic_records = getattr(target_record, polymorphic_relationship) + type_polymorphic_records = [ + r for r in polymorphic_records if getattr(r, type_field) == polymorphic_type + ] + sorted_type_polymorphic_records = sorted( + type_polymorphic_records, key=lambda r: r.start_date, reverse=latest + ) + return sorted_type_polymorphic_records[0] + + if __name__ == "__main__": x = -106.904107 y = 34.068198 diff --git a/tests/features/steps/well-additional-information.py b/tests/features/steps/well-additional-information.py new file mode 100644 index 000000000..02aa4ff22 --- /dev/null +++ b/tests/features/steps/well-additional-information.py @@ -0,0 +1,206 @@ +from behave import when, then + +from services.util import retrieve_polymorphic_table_record + + +@when("the user retrieves the well by ID via path parameter") +def step_impl_retrieve_well_by_id(context): + context.well = context.objects["wells"][0] + context.response = context.client.get(f"/thing/water-well/{context.well.id}") + context.data = context.response.json() + + +# ------------------------------------------------------------------------------ +# Permissions / Operational OK flags +# ------------------------------------------------------------------------------ +# TODO: the API needs to be updated to include Permissions +# TODO: the schema and test data need to be updated +# TODO: should the testing data and tests contain multiple permissions, one that has expired? +# TODO: what are the permission_types that will be used? after they have been determined update these tests + + +@then( + "the response should include whether repeat measurement permission is granted for the well" +) +def step_impl(context): + assert "permissions" in context.data + + permission_record = retrieve_polymorphic_table_record( + context.well, "permissions", "allow_water_level_measurements", latest=True + ) + + assert ( + context.data["permissions"]["allow_water_level_measurements"] + == permission_record.permission_allowed + ) + + +@then("the response should include whether sampling permission is granted for the well") +def step_impl(context): + assert "permissions" in context.data + + permission_record = retrieve_polymorphic_table_record( + context.well, "permissions", "allow_water_chemistry_sample", latest=True + ) + + assert ( + context.data["permissions"]["allow_sampling"] + == permission_record.permission_allowed + ) + + +# TODO: should this be datalogger specific? +@then( + "the response should include whether datalogger installation permission is granted for the well" +) +def step_impl(context): + assert "permissions" in context.data + + permission_record = retrieve_polymorphic_table_record( + context.well, "permissions", "allow_data_logger_installation", latest=True + ) + + assert ( + context.data["permissions"]["allow_data_logger_installation"] + == permission_record.permission_allowed + ) + + +# ------------------------------------------------------------------------------ +# Well Construction Information +# ------------------------------------------------------------------------------ + + +# TODO: needs to be added to model, schemas, test data +@then("the response should include the completion date of the well") +def step_impl(context): + assert "completion_date" in context.data + assert context.data["completion_date"] == context.well.completion_date.strftime( + "%Y-%m-%d" + ) + + +# TODO: needs to be added to model, schemas, test data +@then("the response should include the source of the completion information") +def step_impl(context): + assert "completion_info_source" in context.data + assert context.data["completion_info_source"] == context.well.completion_info_source + + +# TODO: needs to be added to model, schemas, test data +@then("the response should include the driller name") +def step_impl(context): + assert "driller_name" in context.data + assert context.data["driller_name"] == context.well.driller_name + + +# TODO: needs to be added to model, schemas, test data +# TODO: needs to be an enum and added to lexicon +@then("the response should include the construction method") +def step_impl(context): + assert "construction_method" in context.data + assert context.data["construction_method"] == context.well.construction_method + + +# TODO: needs to be added to model, schemas, test data +@then("the response should include the source of the construction information") +def step_impl(context): + assert "construction_info_source" in context.data + assert ( + context.data["construction_info_source"] + == context.well.construction_info_source + ) + + +# ------------------------------------------------------------------------------ +# Additional Well Physical Properties +# ------------------------------------------------------------------------------ + + +# TODO: the transfer script needs to convert ft to in +@then("the response should include the casing diameter in inches") +def step_impl(context): + assert "casing_diameter" in context.data + assert "casing_diameter_unit" in context.data + + assert context.data["casing_diameter"] == context.well.casing_diameter + assert context.data["casing_diameter_unit"] == "in" + + +@then("the response should include the casing depth in feet below ground surface") +def step_impl(context): + assert "well_casing_depth" in context.data + assert "well_casing_depth_unit" in context.data + + assert context.data["well_casing_depth"] == context.well.well_casing_depth + assert context.data["well_casing_depth_unit"] == "ft" + + +# TODO: needs to be added to model, schemas, test data +@then("the response should include the casing materials") +def step_impl(context): + assert "well_casing_materials" in context.data + assert sorted(context.data["well_casing_materials"]) == sorted( + [m.material for m in context.well.well_casing_materials] + ) + + +# TODO: needs to be added to model, schemas, test data +# TODO: needs to be added to lexicon and an enum should be created +@then("the response should include the well pump type (previously well_type field)") +def step_impl(context): + assert "well_pump_type" in context.data + assert context.data["well_pump_type"] == context.well.well_pump_type + + +# TODO: needs to be added to model, schemas, test data +@then("the response should include the well pump depth in feet (new field)") +def step_impl(context): + assert "well_pump_depth" in context.data + assert "well_pump_depth_unit" in context.data + + assert context.data["well_pump_depth"] == context.well.well_pump_depth + assert context.data["well_pump_depth_unit"] == "ft" + + +# TODO: needs to be added to model, schemas, test data +@then( + "the response should include whether the well is open and suitable for a datalogger" +) +def step_impl(context): + data = context.response.json() + assert data["well_open"] is True + assert data["well_suitable_for_datalogger"] is True + + +# ------------------------------------------------------------------------------ +# Aquifer/ Geology Information +# ------------------------------------------------------------------------------ + + +# TODO: needs to be added to model, schemas, test data +@then( + "the response should include the formation as the formation zone of well completion" +) +def step_impl(context): + assert "formation" in context.data + assert context.data["formation"] == context.well.formation + + +# TODO: needs to be added to model, schemas, test data, lexicon +@then( + "the response should include the aquifer class code to classify the aquifer into aquifer system." +) +def step_impl(context): + assert "aquifer_class_code" in context.data + assert context.data["aquifer_class_code"] == context.well.aquifer_class_code + + +# TODO: needs to be added to model, schemas, test data +# TODO: should this be plural? that is, a descriptor model of the well +@then( + "the response should include the aquifer type as the type of aquifers penetrated by the well" +) +def step_impl(context): + assert "aquifer_type" in context.data + assert context.data["aquifer_type"] == context.well.aquifer_type