From d1912ba72844bf0e5bb7ed021e80eab45c5ab83b Mon Sep 17 00:00:00 2001 From: Guillermo Garcia Date: Wed, 29 Jul 2026 17:15:50 +0200 Subject: [PATCH] fix: fixed ecb ids drift between eba dpm releases --- .../services/ecb_validations_import.py | 86 +++++++++- .../migration/test_ecb_validations_import.py | 153 ++++++++++++++++++ 2 files changed, 233 insertions(+), 6 deletions(-) diff --git a/src/dpmcore/services/ecb_validations_import.py b/src/dpmcore/services/ecb_validations_import.py index acf4e47..ba9cc57 100644 --- a/src/dpmcore/services/ecb_validations_import.py +++ b/src/dpmcore/services/ecb_validations_import.py @@ -72,6 +72,13 @@ class EcbValidationsImportResult: class EcbValidationsImportService: """Import ECB validation rules from a CSV export into a DPM database.""" + # Fixed floor for every ID this service assigns, so ECB's IDs stay put + # across releases instead of drifting with the native EBA/DPM dataset. + _ECB_ID_BASE = 1_030_000_000 + + # Required headroom between the native max ID and `_ECB_ID_BASE`. + _ECB_ID_BASE_MIN_MARGIN = 1_000_000 + def __init__(self, engine: Engine) -> None: """Initialise the service with a SQLAlchemy engine.""" self._engine = engine @@ -183,10 +190,57 @@ def _get_parent_operation_code(code: str) -> Optional[str]: return match.group(1) if match else None @staticmethod - def _next_int_id(session: Session, model: Any, attr_name: str) -> int: + def _next_int_id( + session: Session, + model: Any, + attr_name: str, + *, + floor: Optional[int] = None, + ) -> int: + """Return the next available ID for ``attr_name`` on ``model``. + + Without ``floor`` this is a plain "max + 1" over the whole table, + which drifts release to release as the surrounding EBA/DPM dataset + grows. Passing ``floor`` scopes the scan to IDs already inside that + reserved block, so the first assigned ID stays the same run after + run regardless of how large the rest of the table has become. + """ + column = getattr(model, attr_name) + query = session.query(func.max(column)) + if floor is not None: + query = query.filter(column >= floor) + current = query.scalar() + if current is None: + return floor if floor is not None else 1 + return int(current) + 1 + + @staticmethod + def _assert_id_floor_margin( + session: Session, + model: Any, + attr_name: str, + *, + floor: int, + min_margin: int, + ) -> None: + """Fail loudly if native IDs have drifted too close to ``floor``. + + ``floor`` only keeps ECB's IDs stable as long as every *native* + (non-ECB) row in the table stays comfortably below it. Only ECB + import ever writes IDs `>= floor`, so ``max(id) < floor`` is always + the native high-water mark, regardless of whether this run is a + fresh import or a re-run over an already-imported database. + """ column = getattr(model, attr_name) - current = session.query(func.max(column)).scalar() - return int(current or 0) + 1 + native_max = ( + session.query(func.max(column)).filter(column < floor).scalar() + ) + if native_max is not None and floor - int(native_max) < min_margin: + raise EcbValidationsImportError( + f"Native {model.__name__}.{attr_name} max ({native_max})" + f" is within {min_margin} of _ECB_ID_BASE ({floor});" + " bump _ECB_ID_BASE." + ) def _create_operation_concept( self, @@ -383,9 +437,14 @@ def _create_table_references( if not table_codes: return 0 - next_node_id = self._next_int_id(session, OperationNode, "node_id") + next_node_id = self._next_int_id( + session, OperationNode, "node_id", floor=self._ECB_ID_BASE + ) next_ref_id = self._next_int_id( - session, OperandReference, "operand_reference_id" + session, + OperandReference, + "operand_reference_id", + floor=self._ECB_ID_BASE, ) node = OperationNode( @@ -537,14 +596,29 @@ def _import_ecb_validations_df( # noqa: C901 default=None, ) + for model, attr_name in ( + (Operation, "operation_id"), + (OperationVersion, "operation_vid"), + (OperationNode, "node_id"), + (OperandReference, "operand_reference_id"), + ): + self._assert_id_floor_margin( + session, + model, + attr_name, + floor=self._ECB_ID_BASE, + min_margin=self._ECB_ID_BASE_MIN_MARGIN, + ) + counters = { "operation_id": self._next_int_id( - session, Operation, "operation_id" + session, Operation, "operation_id", floor=self._ECB_ID_BASE ), "operation_vid": self._next_int_id( session, OperationVersion, "operation_vid", + floor=self._ECB_ID_BASE, ), } operations_created = 0 diff --git a/tests/unit/migration/test_ecb_validations_import.py b/tests/unit/migration/test_ecb_validations_import.py index 115fc6a..d928244 100644 --- a/tests/unit/migration/test_ecb_validations_import.py +++ b/tests/unit/migration/test_ecb_validations_import.py @@ -238,6 +238,106 @@ def test_next_int_id_returns_max_plus_one( finally: session.close() + def test_next_int_id_with_floor_ignores_rows_below_floor( + self, service_with_schema, sqlite_engine_with_schema + ): + session = sessionmaker(bind=sqlite_engine_with_schema)() + try: + # Rows below the floor (e.g. another org's growing dataset) + # must not influence the floor-scoped block's next ID. + session.add( + Organisation(org_id=5, name="Org5", acronym="O5", id_prefix=1) + ) + session.flush() + result = EcbValidationsImportService._next_int_id( + session, Organisation, "org_id", floor=1000 + ) + assert result == 1000 + finally: + session.close() + + def test_next_int_id_with_floor_returns_max_plus_one_inside_block( + self, service_with_schema, sqlite_engine_with_schema + ): + session = sessionmaker(bind=sqlite_engine_with_schema)() + try: + session.add( + Organisation(org_id=5, name="Org5", acronym="O5", id_prefix=1) + ) + session.add( + Organisation( + org_id=1000, name="Org1000", acronym="O1K", id_prefix=2 + ) + ) + session.flush() + result = EcbValidationsImportService._next_int_id( + session, Organisation, "org_id", floor=1000 + ) + assert result == 1001 + finally: + session.close() + + def test_assert_id_floor_margin_passes_with_sufficient_margin( + self, service_with_schema, sqlite_engine_with_schema + ): + session = sessionmaker(bind=sqlite_engine_with_schema)() + try: + session.add( + Organisation(org_id=5, name="Org5", acronym="O5", id_prefix=1) + ) + session.flush() + EcbValidationsImportService._assert_id_floor_margin( + session, + Organisation, + "org_id", + floor=1000, + min_margin=100, + ) + finally: + session.close() + + def test_assert_id_floor_margin_passes_when_no_native_rows( + self, service_with_schema, sqlite_engine_with_schema + ): + session = sessionmaker(bind=sqlite_engine_with_schema)() + try: + EcbValidationsImportService._assert_id_floor_margin( + session, + Organisation, + "org_id", + floor=1000, + min_margin=100, + ) + finally: + session.close() + + def test_assert_id_floor_margin_raises_when_native_ids_drift_too_close( + self, service_with_schema, sqlite_engine_with_schema + ): + session = sessionmaker(bind=sqlite_engine_with_schema)() + try: + # Native org_id has drifted within the reserved margin of the + # floor: the reservation is no longer a safe anchor. + session.add( + Organisation( + org_id=950, name="Org950", acronym="O950", id_prefix=1 + ) + ) + session.flush() + with pytest.raises( + EcbValidationsImportError, + match="within 100 of _ECB_ID_BASE", + ): + EcbValidationsImportService._assert_id_floor_margin( + session, + Organisation, + "org_id", + floor=1000, + min_margin=100, + ) + finally: + session.close() + def test_get_or_create_ecb_organisation_creates_new( self, service_with_schema, sqlite_engine_with_schema ): @@ -495,6 +595,59 @@ def test_import_csv_duplicate_version_key_skipped( assert result.operation_versions_created == 1 + def test_operation_ids_stable_despite_growing_dataset(self, tmp_path): + """Regression: ECB IDs must not drift as the EBA/DPM dataset grows. + + Before this fix, ECB Operation/OperationVersion IDs were assigned + as ``max(id) + 1`` over the *whole* table, so they silently + shifted release to release as unrelated EBA validations were + added ahead of them. Anchoring to the fixed ``_ECB_ID_BASE`` floor + keeps the assigned ID the same regardless of how large the rest + of the table has grown. + """ + from dpmcore.orm.base import Base + from dpmcore.orm.operations import Operation + + csv_file = tmp_path / "ecb.csv" + csv_file.write_text( + "vr_code,start_release\nV1,4.0\n", encoding="utf-8" + ) + + def _assigned_operation_id(dummy_count: int) -> int: + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + + session = sessionmaker(bind=engine)() + for i in range(1, dummy_count + 1): + session.add( + Operation( + operation_id=i, + code=f"DUMMY_{i}", + type="validation", + source="hierarchy", + ) + ) + session.commit() + session.close() + + EcbValidationsImportService(engine).import_csv(str(csv_file)) + + session = sessionmaker(bind=engine)() + try: + operation = ( + session.query(Operation) + .filter(Operation.code == "V1") + .one() + ) + return operation.operation_id + finally: + session.close() + + small_dataset_id = _assigned_operation_id(10) + large_dataset_id = _assigned_operation_id(5000) + + assert small_dataset_id == large_dataset_id + def test_extract_table_codes_for_empty_expression_returns_empty_set( self, service_with_schema, sqlite_engine_with_schema ):