diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 7087f89..b3ac744 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -307,7 +307,7 @@ def _transform_add_data_resource( endpoints=endpoints if endpoints else None, ) entity_org_mapping = _create_entity_organisation( - new_lookups, dataset, organisations[0] + new_lookups, dataset, organisations[0], pipeline_dir ) # Reload pipeline to pick up newly saved lookups before rerunning transform. @@ -522,36 +522,66 @@ def _get_existing_entities_breakdown(existing_entities): return breakdown -def _create_entity_organisation(new_entities, dataset, organisation): +def _create_entity_organisation( # noqa: C901 + new_entities, dataset, organisation, pipeline_dir +): """ Create entity-organisation mapping from new entities. - Args: - new_entities: List of entity dicts with 'entity' key - dataset: Dataset name - organisation: Organisation identifier + checks whether the new entities already fall within an existing entity-minimum/ + entity-maximum range for this dataset. If the CSV can't be loaded, + processing continues but the returned mapping is flagged with error. - Returns: - List with single dict containing dataset, entity-minimum, entity-maximum, organisation """ if not new_entities: return [] entity_values = [ - entry.get("entity") for entry in new_entities if entry.get("entity") + int(entry.get("entity")) + for entry in new_entities + if entry.get("entity") is not None ] if not entity_values: return [] - return [ - { - "dataset": dataset, - "entity-minimum": min(entity_values), - "entity-maximum": max(entity_values), - "organisation": organisation, - } - ] + entity_org_csv_path = os.path.join(pipeline_dir, "entity-organisation.csv") + try: + with open(entity_org_csv_path, "r", encoding="utf-8") as f: + existing_rows = list(csv.DictReader(f)) + error = False + except (OSError, csv.Error) as err: + logger.warning(f"Unable to load entity-organisation.csv: {err}") + existing_rows = [] + error = True + + overlap = False + if not error: + for row in existing_rows: + if row.get("dataset") != dataset: + continue + try: + row_min = int(row.get("entity-minimum")) + row_max = int(row.get("entity-maximum")) + except (TypeError, ValueError): + continue + if all(row_min <= value <= row_max for value in entity_values): + overlap = True + break + + entry = { + "dataset": dataset, + "organisation": organisation, + "overlap": overlap, + "error": error, + } + # Omit the range when it can't be trusted, so a downstream consumer + # can't blindly commit it to entity-organisation.csv + if not overlap and not error: + entry["entity-minimum"] = min(entity_values) + entry["entity-maximum"] = max(entity_values) + + return [entry] def _map_transformed_entities(transformed_csv_path, pipeline_dir): # noqa: C901 diff --git a/request-processor/tests/unit/src/application/core/test_pipeline.py b/request-processor/tests/unit/src/application/core/test_pipeline.py index 669af4a..8bc740b 100644 --- a/request-processor/tests/unit/src/application/core/test_pipeline.py +++ b/request-processor/tests/unit/src/application/core/test_pipeline.py @@ -6,6 +6,7 @@ fetch_add_data_response, _get_entities_breakdown, _get_existing_entities_breakdown, + _create_entity_organisation, _format_task_summary, _find_duplicate_candidates, run_task_pipeline, @@ -486,6 +487,89 @@ def test_get_entities_breakdown_missing_fields(): assert result[0]["organisation"] == "" +# --- _create_entity_organisation --- + + +def test_create_entity_organisation_sets_overlap_true_when_range_already_present( + tmp_path, +): + """New entities that already fall within an existing range should flag overlap""" + pipeline_dir = tmp_path / "pipeline" + pipeline_dir.mkdir() + (pipeline_dir / "entity-organisation.csv").write_text( + "dataset,entity-minimum,entity-maximum,organisation\n" + "nature-improvement-area,10100000,10100011,government-organisation:PB202\n" + ) + + new_entities = [{"entity": "10100002"}, {"entity": "10100005"}] + + result = _create_entity_organisation( + new_entities, + "nature-improvement-area", + "government-organisation:PB202", + str(pipeline_dir), + ) + + assert len(result) == 1 + assert result[0]["overlap"] is True + assert result[0]["error"] is False + assert "entity-minimum" not in result[0] + assert "entity-maximum" not in result[0] + + +def test_create_entity_organisation_no_overlap_when_range_not_present(tmp_path): + """New entities outside all existing ranges should not flag overlap""" + pipeline_dir = tmp_path / "pipeline" + pipeline_dir.mkdir() + (pipeline_dir / "entity-organisation.csv").write_text( + "dataset,entity-minimum,entity-maximum,organisation\n" + "nature-improvement-area,10100000,10100011,government-organisation:PB202\n" + ) + + new_entities = [{"entity": "10200000"}, {"entity": "10200001"}] + + result = _create_entity_organisation( + new_entities, + "nature-improvement-area", + "government-organisation:PB202", + str(pipeline_dir), + ) + + assert len(result) == 1 + assert result[0]["overlap"] is False + assert result[0]["error"] is False + assert result[0]["entity-minimum"] == 10200000 + assert result[0]["entity-maximum"] == 10200001 + + +def test_create_entity_organisation_missing_csv_sets_error_true(tmp_path): + """Missing entity-organisation.csv should set error but still return a mapping""" + pipeline_dir = tmp_path / "pipeline" + pipeline_dir.mkdir() + + new_entities = [{"entity": "10100002"}] + + result = _create_entity_organisation( + new_entities, + "nature-improvement-area", + "government-organisation:PB202", + str(pipeline_dir), + ) + + assert len(result) == 1 + assert result[0]["error"] is True + assert result[0]["overlap"] is False + assert "entity-minimum" not in result[0] + assert "entity-maximum" not in result[0] + + +def test_create_entity_organisation_empty_entities_returns_empty_list(tmp_path): + result = _create_entity_organisation( + [], "nature-improvement-area", "government-organisation:PB202", str(tmp_path) + ) + assert result == [] + + def test_get_existing_entities_breakdown_success(): """Test converting existing entities to simplified format""" existing_entities = [