From bb89eaed91b9b60ef21ac6e723c27458fd23b91e Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 9 Sep 2026 12:37:00 -0400 Subject: [PATCH 1/2] Derive mirrored policy identity from persisted JSON --- .../persisted-policy-mirror-identity.fixed.md | 1 + policyengine_api/services/policy_service.py | 5 + tests/unit/services/test_policy_service.py | 25 +++++ tests/unit/v2/test_policy_canonicalization.py | 46 +++++++++ .../unit/v2/test_policy_legacy_translation.py | 98 +++++++++++++++++++ 5 files changed, 175 insertions(+) create mode 100644 changelog.d/persisted-policy-mirror-identity.fixed.md diff --git a/changelog.d/persisted-policy-mirror-identity.fixed.md b/changelog.d/persisted-policy-mirror-identity.fixed.md new file mode 100644 index 000000000..d933c3058 --- /dev/null +++ b/changelog.d/persisted-policy-mirror-identity.fixed.md @@ -0,0 +1 @@ +Build a new policy's mirror identity from its persisted JSON so database numeric normalization cannot give equivalent new requests and retries different identities. diff --git a/policyengine_api/services/policy_service.py b/policyengine_api/services/policy_service.py index b8b011f43..4dca63791 100644 --- a/policyengine_api/services/policy_service.py +++ b/policyengine_api/services/policy_service.py @@ -152,6 +152,11 @@ def set_policy( policy_json, policy_hash, ) + if prepare_for_mirroring and not is_existing_policy: + # MySQL JSON storage can change numeric representation. Mirror + # the stored content on creation, just as an existing-row retry + # does, so both attempts use the same v2 content identity. + session.refresh(policy) snapshot = ( LegacyPolicySnapshot( country_id=policy.country_id, diff --git a/tests/unit/services/test_policy_service.py b/tests/unit/services/test_policy_service.py index c8f008dd7..2bb09ed8e 100644 --- a/tests/unit/services/test_policy_service.py +++ b/tests/unit/services/test_policy_service.py @@ -156,6 +156,11 @@ def test_set_policy_does_not_build_v2_snapshot_unless_requested( "policyengine_api.services.policy_service.hash_object", lambda value: "new-hash", ) + monkeypatch.setattr( + service._sessions.class_, + "refresh", + lambda *args, **kwargs: pytest.fail("cloud_sql-only writes must not refresh"), + ) result = service.set_policy( "ca", @@ -189,3 +194,23 @@ def test_set_policy_propagates_flush_failure( with pytest.raises(SQLAlchemyError, match="insert failed"): service.set_policy("us", "Policy", {}) + + +def test_mirror_snapshot_read_failure_rolls_back_new_v1_policy( + service, orm_session_factory, monkeypatch +): + with monkeypatch.context() as failing_read: + failing_read.setattr( + orm_session_factory.class_, + "refresh", + lambda *args, **kwargs: (_ for _ in ()).throw( + SQLAlchemyError("snapshot read failed") + ), + ) + with pytest.raises(SQLAlchemyError, match="snapshot read failed"): + service.set_policy("us", "Failed snapshot", {}, prepare_for_mirroring=True) + + assert service.search_policies("us", "Failed snapshot") == [] + retry = service.set_policy("us", "Failed snapshot", {}, prepare_for_mirroring=True) + assert retry.is_existing_policy is False + assert retry.snapshot == service.get_policy_snapshot("us", retry.policy_id) diff --git a/tests/unit/v2/test_policy_canonicalization.py b/tests/unit/v2/test_policy_canonicalization.py index 3b750456b..8d0e9ffcd 100644 --- a/tests/unit/v2/test_policy_canonicalization.py +++ b/tests/unit/v2/test_policy_canonicalization.py @@ -4,8 +4,11 @@ from datetime import datetime, timezone import hashlib +import math from uuid import UUID, uuid4 +import pytest + from policyengine_api.services.v2.policies.transformations import ( POLICY_CANONICALIZATION_VERSION, canonical_policy_document, @@ -104,6 +107,49 @@ def test_equivalent_json_numbers_and_utc_instants_have_one_encoding() -> None: assert canonicalize_policy(integer) == canonicalize_policy(floating) +@pytest.mark.parametrize( + ("first_value", "second_value"), + [ + pytest.param(0.04, math.nextafter(0.04, math.inf), id="live-probe-lower-rate"), + pytest.param( + 0.044999, + math.nextafter(0.044999, math.inf), + id="live-probe-upper-rate", + ), + pytest.param(1.0, math.nextafter(1.0, math.inf), id="adjacent-to-integer"), + pytest.param(1e-12, math.nextafter(1e-12, math.inf), id="small-float"), + pytest.param(True, 1, id="true-is-not-one"), + pytest.param(False, 0, id="false-is-not-zero"), + pytest.param(2**53, 2**53 + 1, id="integers-beyond-float-precision"), + pytest.param( + {"rates": [0.04, {"enabled": True}]}, + {"rates": [math.nextafter(0.04, math.inf), {"enabled": True}]}, + id="nested-adjacent-floats", + ), + ], +) +def test_distinct_numeric_content_retains_distinct_identity( + first_value: object, second_value: object +) -> None: + contents = [ + canonicalize_policy( + _command( + values=[ + { + "parameter_id": FIRST_PARAMETER_ID, + "value": value, + "start_date": "2026-01-01T00:00:00Z", + } + ] + ) + ) + for value in (first_value, second_value) + ] + + assert contents[0].document != contents[1].document + assert contents[0].content_hash != contents[1].content_hash + + def test_material_content_changes_produce_distinct_documents() -> None: original = canonical_policy_document(_command()) alternatives = [ diff --git a/tests/unit/v2/test_policy_legacy_translation.py b/tests/unit/v2/test_policy_legacy_translation.py index bf569616e..8bb58e663 100644 --- a/tests/unit/v2/test_policy_legacy_translation.py +++ b/tests/unit/v2/test_policy_legacy_translation.py @@ -3,8 +3,10 @@ from __future__ import annotations from datetime import datetime, timezone +import json from pydantic import ValidationError +from sqlalchemy.orm import sessionmaker from sqlmodel import Session, create_engine import pytest @@ -19,6 +21,7 @@ read_policy_catalog, ) from policyengine_api.services.v2.policies.transformations import ( + canonicalize_policy, parse_legacy_period, translate_legacy_policy, ) @@ -26,6 +29,12 @@ from policyengine_api.services.v2.policies.validators import ( LegacyPolicyTranslationError, ) +from policyengine_api.services.policy_service import PolicyService +from policyengine_api.services.policy_mirroring import ( + PolicyMirrorUnavailableError, + mirror_policy_after_commit, +) +from tests.fixtures.local_v1_database import create_test_v1_schema def _session_and_catalog(): @@ -137,6 +146,95 @@ def test_label_does_not_change_translated_core_content() -> None: engine.dispose() +@pytest.mark.parametrize( + ("request_value", "stored_value"), + [ + (0.04 + 3 / 1_000_000, 0.040003), + (1.4334335999999999, 1.4334336), + (-5.684341886080803e-14, -5.684341886080804e-14), + ], +) +def test_new_retry_and_equivalent_label_mirror_the_same_stored_json( + monkeypatch, request_value: float, stored_value: float +) -> None: + # MySQL's default RapidJSON parser can move a double by one ULP: + # https://bugs.mysql.com/bug.php?id=112904 + # The first pair also comes from the live Phase 10 probe's value formula. + # Model those specific storage round trips at the database boundary. Plain + # SQLite JSON would preserve the request float and hide this regression. + def serialize_stored_json(value): + return json.dumps( + json.loads( + json.dumps(value), + parse_float=lambda number: ( + stored_value if number == repr(request_value) else float(number) + ), + ) + ) + + v1_engine = create_engine("sqlite://", json_serializer=serialize_stored_json) + create_test_v1_schema(v1_engine) + service = PolicyService(sessionmaker(v1_engine, expire_on_commit=False)) + monkeypatch.setattr( + "policyengine_api.services.policy_service.COUNTRY_PACKAGE_VERSIONS", + {"us": "1.0.0"}, + ) + engine, session, _model, _version, _first, _second = _session_and_catalog() + request = { + "gov.example.rate": {"2026-01-01.2100-12-31": request_value}, + "gov.example.amount": {"2026": 100}, + } + try: + first = service.set_policy( + "us", "First label", request, prepare_for_mirroring=True + ) + + def unavailable_mirror(): + raise RuntimeError("controlled destination failure") + + with pytest.raises(PolicyMirrorUnavailableError): + mirror_policy_after_commit( + first.snapshot, mirror_factory=unavailable_mirror + ) + assert service.get_policy("us", first.policy_id) is not None + # Retry loads the committed row; the equivalent-label request creates + # a different legacy row. + retry = service.set_policy( + "us", "First label", request, prepare_for_mirroring=True + ) + equivalent = service.set_policy( + "us", "Equivalent label", request, prepare_for_mirroring=True + ) + assert first.is_existing_policy is False + assert retry.is_existing_policy is True + assert equivalent.is_existing_policy is False + assert retry.policy_id == first.policy_id + assert equivalent.policy_id != first.policy_id + assert equivalent.snapshot.label != first.snapshot.label + assert ( + equivalent.snapshot.source_policy_hash == first.snapshot.source_policy_hash + ) + contents = [ + canonicalize_policy(_translate(session, result.snapshot)) + for result in (first, retry, equivalent) + ] + assert contents[2] == contents[1] + assert contents[0] == contents[1] + stored = service.get_policy_snapshot("us", first.policy_id) + assert stored.policy_json["gov.example.rate"] == { + "2026-01-01.2100-12-31": stored_value + } + assert all( + result.snapshot.policy_json == stored.policy_json + for result in (first, retry, equivalent) + ) + assert request["gov.example.rate"]["2026-01-01.2100-12-31"] == request_value + finally: + session.close() + engine.dispose() + v1_engine.dispose() + + @pytest.mark.parametrize( "policy_json", [ From 6d8ab721f93911f8fa27f2ddaeb8bdf27799323a Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 9 Sep 2026 15:31:47 -0400 Subject: [PATCH 2/2] Qualify policy mirroring against real MySQL storage in CI --- .github/workflows/v2-integration-check.yml | 18 +- .../test_mysql_policy_dual_write.py | 424 ++++++++++++++++++ 2 files changed, 441 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_mysql_policy_dual_write.py diff --git a/.github/workflows/v2-integration-check.yml b/.github/workflows/v2-integration-check.yml index 40e362171..c757302ae 100644 --- a/.github/workflows/v2-integration-check.yml +++ b/.github/workflows/v2-integration-check.yml @@ -12,6 +12,19 @@ jobs: name: V2 PostgreSQL and Redis integration runs-on: ubuntu-latest services: + mysql: + image: mysql:8.4 + # Disposable job-owned credentials, matching the v1 lifecycle target. + env: + MYSQL_ROOT_PASSWORD: policyengine_test + MYSQL_DATABASE: policyengine_alembic_test + ports: + - 3306:3306 + options: >- + --health-cmd="mysqladmin ping --silent" + --health-interval=5s + --health-timeout=5s + --health-retries=20 postgres: image: postgres:17 env: @@ -35,6 +48,7 @@ jobs: --health-timeout=5s --health-retries=20 env: + ALEMBIC_DATABASE_URL: mysql+pymysql://root:policyengine_test@127.0.0.1:3306/policyengine_alembic_test V2_MIGRATION_DATABASE_URL: postgresql+psycopg://postgres:policyengine_v2_test@127.0.0.1:5432/policyengine_v2_alembic_test V2_ALEMBIC_DISPOSABLE_TEST: "1" RUNTIME_CACHE_TEST_URL: redis://127.0.0.1:6379/0 @@ -51,6 +65,8 @@ jobs: run: uv sync --frozen - name: Prepare the disposable v2 schema run: uv run alembic -c alembic-v2.ini upgrade head + - name: Prepare the disposable v1 MySQL schema + run: uv run alembic -c alembic-v1.ini upgrade head - name: Verify the installed PolicyEngine.py catalog interface run: uv run coverage run --branch -m pytest -q tests/integration/test_v2_catalog_installed.py env: @@ -58,7 +74,7 @@ jobs: - name: Test v2 metadata publication and resource routes run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_catalog_publication.py tests/integration/test_v2_metadata_routes.py - name: Test v2 policy persistence and immediate v1 mirroring - run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_policy_persistence.py tests/integration/test_v1_policy_dual_write.py tests/integration/test_v2_user_policy_mirroring.py tests/integration/test_v1_user_policy_dual_write.py + run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_policy_persistence.py tests/integration/test_v1_policy_dual_write.py tests/integration/test_mysql_policy_dual_write.py tests/integration/test_v2_user_policy_mirroring.py tests/integration/test_v1_user_policy_dual_write.py - name: Qualify production-scale v2 metadata publication run: uv run coverage run -a --branch -m pytest -q tests/integration/test_v2_catalog_publication_qualification.py env: diff --git a/tests/integration/test_mysql_policy_dual_write.py b/tests/integration/test_mysql_policy_dual_write.py new file mode 100644 index 000000000..e3a778192 --- /dev/null +++ b/tests/integration/test_mysql_policy_dual_write.py @@ -0,0 +1,424 @@ +"""Real MySQL JSON storage and PostgreSQL policy-mirroring regressions. + +Opt in with ALEMBIC_DATABASE_URL pointing at local policyengine_alembic_test +and, for cross-database tests, V2_ALEMBIC_DISPOSABLE_TEST=1 plus +V2_MIGRATION_DATABASE_URL pointing at local policyengine_v2_alembic_test. +Prepare both schemas with their respective Alembic upgrades before running. +Only rows created by these fixtures are removed during cleanup. +""" + +from __future__ import annotations + +import json +import os +from contextlib import contextmanager +from dataclasses import dataclass +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy import create_engine, delete, func, select, text +from sqlalchemy.engine import Engine, make_url +from sqlalchemy.exc import DataError +from sqlalchemy.orm import sessionmaker +from sqlmodel import Session + +from policyengine_api.constants import POLICYENGINE_VERSION +from policyengine_api.data.v1_models import Policy as V1Policy +from policyengine_api.data.v2.migration_target import ( + V2_ALEMBIC_DISPOSABLE_TEST, + load_v2_alembic_settings, +) +from policyengine_api.data.v2.models import ( + LegacyPolicyMapping, + Parameter, + ParameterValue, + Policy, + TaxBenefitModel, + TaxBenefitModelVersion, +) +from policyengine_api.data.v2.settings import V2_MIGRATION_DATABASE_URL +from policyengine_api.services.policy_mirroring import ( + PolicyMirrorUnavailableError, + mirror_policy_after_commit, +) +from policyengine_api.services.policy_service import PolicyService, PolicySetResult +from policyengine_api.services.v2.policies.database_session import PolicyDatabaseSession +from policyengine_api.services.v2.policies.services import V2PolicyService +from policyengine_api.services.v2.policies.types import ( + NativePolicyCreationInput, + PolicyParameterValueInput, +) +from policyengine_api.services.v2.policies.validators import ( + LegacyPolicyMappingIntegrityError, +) +from policyengine_api.utils import hash_object + +INPUT_VALUE = 0.04 + 3 / 1_000_000 +STORED_VALUE = 0.040003 +PARAMETER_NAME = "gov.phase10.mysql_storage_rate" +PERIOD = "2026-01-01.2100-12-31" + + +def _mysql_url() -> str: + database_url = os.environ.get("ALEMBIC_DATABASE_URL", "") + if not database_url: + pytest.skip("ALEMBIC_DATABASE_URL is not set") + url = make_url(database_url) + if ( + url.drivername != "mysql+pymysql" + or url.host not in {"127.0.0.1", "localhost"} + or url.database != "policyengine_alembic_test" + or not url.username + or url.password is None + ): + pytest.fail( + "MySQL mirror tests require explicit test credentials and the local " + "mysql+pymysql policyengine_alembic_test schema" + ) + return database_url + + +def _postgres_url() -> str: + database_url = os.environ.get(V2_MIGRATION_DATABASE_URL, "") + if not database_url: + pytest.skip(f"{V2_MIGRATION_DATABASE_URL} is not set") + if os.environ.get(V2_ALEMBIC_DISPOSABLE_TEST) != "1": + pytest.fail("MySQL mirror tests require v2 disposable-test mode") + settings = load_v2_alembic_settings( + { + V2_MIGRATION_DATABASE_URL: database_url, + V2_ALEMBIC_DISPOSABLE_TEST: "1", + } + ) + return settings.url.render_as_string(hide_password=False) + + +@dataclass +class MySQLSource: + engine: Engine + service: PolicyService + label_prefix: str + + def create(self, label: str = "first") -> PolicySetResult: + return self.service.set_policy( + "us", + f"{self.label_prefix}{label}", + {PARAMETER_NAME: {PERIOD: INPUT_VALUE}}, + prepare_for_mirroring=True, + ) + + +@pytest.fixture +def mysql_source(): + engine = create_engine(_mysql_url()) + source = MySQLSource( + engine, + PolicyService(sessionmaker(engine, expire_on_commit=False)), + f"mysql-mirror-{uuid4().hex}-", + ) + try: + yield source + finally: + try: + with engine.begin() as connection: + connection.execute( + delete(V1Policy).where( + V1Policy.label.startswith(source.label_prefix, autoescape=True) + ) + ) + finally: + engine.dispose() + + +@dataclass +class PostgreSQLDestination: + sessions: sessionmaker + service: V2PolicyService + model_id: UUID + + def mirror(self, creation: PolicySetResult): + assert creation.snapshot is not None + return mirror_policy_after_commit( + creation.snapshot, + mirror_factory=lambda: self.service, + ) + + def counts(self) -> tuple[int, int, int]: + policy_ids = select(Policy.id).where( + Policy.tax_benefit_model_id == self.model_id + ) + with self.sessions() as session: + return tuple( + session.scalar(select(func.count()).select_from(table).where(clause)) + for table, clause in ( + (Policy, Policy.id.in_(policy_ids)), + (ParameterValue, ParameterValue.policy_id.in_(policy_ids)), + ( + LegacyPolicyMapping, + LegacyPolicyMapping.policy_id.in_(policy_ids), + ), + ) + ) + + +@pytest.fixture +def postgres_destination(): + engine = create_engine(_postgres_url()) + sessions = sessionmaker(engine, class_=Session, expire_on_commit=False) + model_id = None + try: + with sessions.begin() as session: + model = TaxBenefitModel(name="policyengine-us") + version = TaxBenefitModelVersion( + model=model, + version=POLICYENGINE_VERSION, + current_law_id=1, + metadata_time_periods=[2026], + ) + session.add( + Parameter( + name=PARAMETER_NAME, + tax_benefit_model_version=version, + ) + ) + session.flush() + model_id = model.id + yield PostgreSQLDestination( + sessions, V2PolicyService(PolicyDatabaseSession(sessions)), model_id + ) + finally: + try: + if model_id is not None: + with engine.begin() as connection: + policy_ids = select(Policy.id).where( + Policy.tax_benefit_model_id == model_id + ) + version_ids = select(TaxBenefitModelVersion.id).where( + TaxBenefitModelVersion.model_id == model_id + ) + for table, clause in ( + ( + LegacyPolicyMapping, + LegacyPolicyMapping.policy_id.in_(policy_ids), + ), + (ParameterValue, ParameterValue.policy_id.in_(policy_ids)), + (Policy, Policy.tax_benefit_model_id == model_id), + ( + Parameter, + Parameter.tax_benefit_model_version_id.in_(version_ids), + ), + ( + TaxBenefitModelVersion, + TaxBenefitModelVersion.model_id == model_id, + ), + (TaxBenefitModel, TaxBenefitModel.id == model_id), + ): + connection.execute(delete(table).where(clause)) + finally: + engine.dispose() + + +def test_first_snapshot_matches_real_mysql_json_storage( + mysql_source, record_property +) -> None: + creation = mysql_source.create() + assert creation.snapshot is not None + with mysql_source.engine.connect() as connection: + raw_json = connection.execute( + text("SELECT policy_json FROM policy WHERE id = :id AND country_id = 'us'"), + {"id": creation.policy_id}, + ).scalar_one() + persisted = json.loads(raw_json) + record_property("input_value", repr(INPUT_VALUE)) + record_property("mysql_policy_json", raw_json) + record_property("mysql_legacy_policy_id", creation.policy_id) + + # This assertion makes the regression sensitive to actual MySQL storage; + # a SQLite replacement or a backend preserving the input is insufficient. + assert repr(INPUT_VALUE) == "0.040003000000000004" + assert persisted == {PARAMETER_NAME: {PERIOD: STORED_VALUE}} + assert persisted[PARAMETER_NAME][PERIOD] != INPUT_VALUE + assert creation.snapshot.policy_json == persisted + assert creation.snapshot.source_policy_hash == hash_object( + {PARAMETER_NAME: {PERIOD: INPUT_VALUE}} + ) + assert creation.snapshot == mysql_source.service.get_policy_snapshot( + "us", creation.policy_id + ) + + +@pytest.mark.parametrize("fail_destination", [False, True], ids=["success", "rollback"]) +def test_mysql_first_write_retry_and_relabel_keep_one_postgres_uuid( + mysql_source, postgres_destination, fail_destination, record_property +) -> None: + creation = mysql_source.create() + assert creation.is_existing_policy is False + assert creation.snapshot == mysql_source.service.get_policy_snapshot( + "us", creation.policy_id + ) + assert creation.snapshot.policy_json == {PARAMETER_NAME: {PERIOD: STORED_VALUE}} + + if fail_destination: + + class FailingTransaction(PolicyDatabaseSession): + @contextmanager + def transaction(self): + with super().transaction() as session: + yield session + # Fail on the real server after policy/value/mapping inserts + # and before COMMIT, exercising PostgreSQL rollback itself. + session.execute(text("SELECT 1 / 0")) + + failing_service = V2PolicyService( + FailingTransaction(postgres_destination.sessions) + ) + with pytest.raises(PolicyMirrorUnavailableError) as failure: + mirror_policy_after_commit( + creation.snapshot, mirror_factory=lambda: failing_service + ) + assert isinstance(failure.value.__cause__, DataError) + assert postgres_destination.counts() == (0, 0, 0) + assert mysql_source.service.get_policy("us", creation.policy_id) is not None + first = None + else: + first = postgres_destination.mirror(creation) + assert first.policy_created is True + assert first.mapping_created is True + + # Repeat the real source service call, including a new MySQL session, as an + # HTTP retry would after destination failure or a lost successful response. + retry_creation = mysql_source.create() + assert retry_creation.is_existing_policy is True + assert retry_creation.policy_id == creation.policy_id + assert retry_creation.snapshot == creation.snapshot + retry = postgres_destination.mirror(retry_creation) + assert retry.policy_created is fail_destination + assert retry.mapping_created is fail_destination + if first is not None: + assert retry.policy_id == first.policy_id + + relabeled = mysql_source.create("relabeled") + assert relabeled.is_existing_policy is False + assert relabeled.policy_id != creation.policy_id + assert relabeled.snapshot.policy_json == creation.snapshot.policy_json + assert relabeled.snapshot.source_policy_hash == creation.snapshot.source_policy_hash + relabel_result = postgres_destination.mirror(relabeled) + record_property("mysql_first_legacy_policy_id", creation.policy_id) + record_property("mysql_retry_legacy_policy_id", retry_creation.policy_id) + record_property("mysql_relabel_legacy_policy_id", relabeled.policy_id) + record_property( + "postgres_first_policy_id", + "rolled_back" if first is None else str(first.policy_id), + ) + record_property("postgres_retry_policy_id", str(retry.policy_id)) + record_property("postgres_relabel_policy_id", str(relabel_result.policy_id)) + assert relabel_result.policy_id == retry.policy_id + assert relabel_result.policy_created is False + assert relabel_result.mapping_created is True + assert ( + postgres_destination.mirror(mysql_source.create()).policy_id == retry.policy_id + ) + assert ( + postgres_destination.mirror(mysql_source.create("relabeled")).policy_id + == retry.policy_id + ) + assert postgres_destination.counts() == (1, 1, 2) + with postgres_destination.sessions() as session: + mappings = session.scalars( + select(LegacyPolicyMapping).where( + LegacyPolicyMapping.policy_id == retry.policy_id + ) + ).all() + assert {row.legacy_policy_id for row in mappings} == { + creation.policy_id, + relabeled.policy_id, + } + assert {row.source_policy_hash for row in mappings} == { + creation.snapshot.source_policy_hash + } + + +def test_native_postgres_policies_preserve_the_distinct_input_number( + mysql_source, postgres_destination, record_property +) -> None: + mirrored = postgres_destination.mirror(mysql_source.create()) + item = postgres_destination.service.get_policy( + country_id="us", policy_id=mirrored.policy_id + ) + parameter = item.parameter_values[0] + assert parameter.value == STORED_VALUE + + def create_native(value): + return postgres_destination.service.create_policy( + NativePolicyCreationInput( + country_id="us", + tax_benefit_model_id=postgres_destination.model_id, + parameter_values=[ + PolicyParameterValueInput( + parameter_id=parameter.parameter_id, + value=value, + start_date=parameter.start_date, + end_date=parameter.end_date, + ) + ], + ) + ) + + stored = create_native(STORED_VALUE) + distinct = create_native(INPUT_VALUE) + record_property("stored_value", repr(STORED_VALUE)) + record_property("distinct_input_value", repr(INPUT_VALUE)) + record_property("postgres_stored_policy_id", str(stored.item.id)) + record_property("postgres_distinct_policy_id", str(distinct.item.id)) + assert stored.created is False + assert stored.item.id == mirrored.policy_id + assert distinct.created is True + assert distinct.item.id != mirrored.policy_id + assert distinct.item.parameter_values[0].value == INPUT_VALUE + assert create_native(INPUT_VALUE).item.id == distinct.item.id + assert postgres_destination.counts() == (2, 2, 1) + + +def test_historical_pre_storage_mapping_is_rejected_without_repair( + mysql_source, postgres_destination, record_property +) -> None: + creation = mysql_source.create() + assert creation.snapshot is not None + assert creation.snapshot.policy_json == {PARAMETER_NAME: {PERIOD: STORED_VALUE}} + # Reproduce the old first-write snapshot in disposable PostgreSQL only. + historical = creation.snapshot.model_copy( + update={"policy_json": {PARAMETER_NAME: {PERIOD: INPUT_VALUE}}} + ) + original = mirror_policy_after_commit( + historical, mirror_factory=lambda: postgres_destination.service + ) + retry = mysql_source.create() + assert retry.snapshot.source_policy_hash == historical.source_policy_hash + assert retry.snapshot.legacy_policy_id == historical.legacy_policy_id + assert retry.snapshot.policy_json != historical.policy_json + + with pytest.raises(PolicyMirrorUnavailableError) as failure: + postgres_destination.mirror(retry) + assert isinstance(failure.value.__cause__, LegacyPolicyMappingIntegrityError) + assert "translated immutable content" in str(failure.value.__cause__) + record_property("mysql_legacy_policy_id", creation.policy_id) + record_property("postgres_historical_policy_id", str(original.policy_id)) + record_property("retry_error", type(failure.value.__cause__).__name__) + assert postgres_destination.counts() == (1, 1, 1) + with postgres_destination.sessions() as session: + mapping = session.scalar( + select(LegacyPolicyMapping).where( + LegacyPolicyMapping.country_id == "us", + LegacyPolicyMapping.legacy_policy_id == creation.policy_id, + ) + ) + assert mapping.policy_id == original.policy_id + assert mapping.source_policy_hash == historical.source_policy_hash + assert ( + session.scalar( + select(ParameterValue.value_json).where( + ParameterValue.policy_id == original.policy_id + ) + ) + == INPUT_VALUE + )