diff --git a/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md new file mode 100644 index 000000000..31df2fc62 --- /dev/null +++ b/changelog.d/v2plan-1k-142-copy-reused-policies.fixed.md @@ -0,0 +1 @@ +Dual-write now copies a v1 policy to v2 whatever country package version it was saved under, as long as every parameter it sets exists in the running catalog. Before, every policy saved under an older release was refused, which included each `POST /{country}/policy` that matched an existing policy. A v1 policy that already has a v2 copy now returns that copy without being translated again, so writing it again after a PolicyEngine.py or country package upgrade no longer fails. Each failed policy or saved-policy copy now logs a fixed `failure_reason` next to its `failure_category`, such as `missing_parameters`. diff --git a/docs/migration/stage-10-v2-policies.md b/docs/migration/stage-10-v2-policies.md index 252bc940e..e464e9949 100644 --- a/docs/migration/stage-10-v2-policies.md +++ b/docs/migration/stage-10-v2-policies.md @@ -129,7 +129,12 @@ failure, retained-event processing, idempotent retries, and source/destination rollback using the same MySQL and PostgreSQL engines as the deployed databases. Under this selection, a core-policy mutation commits Cloud SQL first and then -completes its policy transaction in Supabase. A saved-policy mutation commits +completes its policy transaction in Supabase. The copy binds the v1 policy to +the running catalog whatever country package version the v1 row was saved +under (its stored `api_version`), the same way the Public API simulates an old +policy with the running country package. A policy is copied when every +parameter path it sets exists in the running catalog and its periods and values +are valid. A saved-policy mutation commits its source row, incremented revision, and complete event in one Cloud SQL transaction. The same request then processes that source's pending events in revision order and records `processed_at` only after the corresponding @@ -147,7 +152,12 @@ its `failure_category`, and the route returns its normal v1 response. A saved-policy event that failed stays pending, and the next mutation of the same saved policy replays it, with earlier retained events first. A core policy whose copy failed is copied again the next time any request writes the same -policy. Destination revision and fingerprint checks make a replay safe if +policy. A core policy that already has a copy returns that copy without being +translated again, so writing it again after a PolicyEngine.py or country +package upgrade does not fail. When two first copies of the same v1 policy run +at once on revisions with different PolicyEngine.py versions, the mapping +committed first is the copy; the other request's v2 policy commits without a +mapping and its record reports `policy_created=false`. Destination revision and fingerprint checks make a replay safe if Supabase committed but the Cloud SQL processing marker did not. There is no background processor or reconciliation process. @@ -168,7 +178,19 @@ functions do not implement separate preflight behavior. Monitor structured events with metric names `v1_policy_mirror_operations` and `v1_user_policy_mirror_operations`. Alert on `outcome=error`, grouped by -`failure_category` and country. Events include the legacy integer ID, +`failure_category` and country. Each failed copy also records a fixed +`failure_reason`: + +| `failure_category` | `failure_reason` | +| --- | --- | +| `catalog_or_translation` | `missing_parameters` (the policy sets a parameter the running catalog lacks), `invalid_parameter_values`, `empty_parameter_values`, `invalid_period`, `conflicting_values`, `too_many_values` (more than 1,000 dated values), `catalog_absent` (no catalog for the running PolicyEngine.py version), `catalog_query_failed`, `catalog_version_not_found`, `catalog_mismatch` | +| `integrity` | `mapping_conflict`, `content_hash_collision`, `stored_policy_integrity`, `saved_policy_integrity` (any integrity check of a saved-policy copy) | +| `database` | `database_error` | +| `configuration` | `configuration` | +| `unexpected` | `missing_reform_policy` (a saved policy's v1 reform could not be loaded), `unexpected` | +| `missing_mirror_snapshot`, `missing_mirror_revision` | the same value as the category | + +Events include the legacy integer ID, destination UUID when committed, attempted and completed database sources, and duration. They do not include policy JSON, presentation data, database URLs, or credentials. @@ -177,7 +199,8 @@ The disposable and route-level tests verify that: - native policy and association create/read/list/update/delete operations use the initialized catalog and Supabase only; -- both newly created and existing v1 policies receive durable mappings; +- both newly created and existing v1 policies receive durable mappings, + including existing policies saved under an older country package version; - v1 saved-policy label updates change association `name`, while v1-only field updates change only the mapping fingerprint; - saved-policy event revisions are processed in ascending order, successful diff --git a/policyengine_api/routes/policy_routes.py b/policyengine_api/routes/policy_routes.py index 1eef717b2..51694175c 100644 --- a/policyengine_api/routes/policy_routes.py +++ b/policyengine_api/routes/policy_routes.py @@ -21,6 +21,10 @@ from policyengine_api.request_context import current_request_id from policyengine_api.response_factory import _make_error_response from policyengine_api.services.policy_mirroring import ( + MISSING_MIRROR_REVISION_FAILURE, + MISSING_MIRROR_SNAPSHOT_FAILURE, + UNEXPECTED_FAILURE, + PolicyCopyFailure, PolicyMirrorUnavailableError, mirror_policy_after_commit, ) @@ -68,7 +72,7 @@ def _log_uncopied_v1_write( legacy_id_field: str, legacy_id: int, country_id: str, - failure_category: str, + failure: PolicyCopyFailure, ) -> None: """Record a committed v1 write that has no v2 copy, without failing the request. @@ -89,7 +93,8 @@ def _log_uncopied_v1_write( legacy_id_field: _safe_legacy_integer_id(legacy_id), "request_id": current_request_id(), "outcome": "error", - "failure_category": failure_category, + "failure_category": failure.category.value, + "failure_reason": failure.reason.value, }, severity="ERROR", ) @@ -111,7 +116,7 @@ def _copy_policy_to_v2( legacy_id_field="legacy_policy_id", legacy_id=policy_id, country_id=country_id, - failure_category="missing_mirror_snapshot", + failure=MISSING_MIRROR_SNAPSHOT_FAILURE, ) return # mirror_policy_after_commit has already logged the failure and its category. @@ -135,7 +140,7 @@ def _copy_saved_policy_events_to_v2( legacy_id_field="legacy_user_policy_id", legacy_id=legacy_user_policy_id, country_id=country_id, - failure_category="missing_mirror_revision", + failure=MISSING_MIRROR_REVISION_FAILURE, ) return try: @@ -157,7 +162,7 @@ def _copy_saved_policy_events_to_v2( legacy_id_field="legacy_user_policy_id", legacy_id=legacy_user_policy_id, country_id=country_id, - failure_category="unexpected", + failure=UNEXPECTED_FAILURE, ) diff --git a/policyengine_api/services/policy_mirroring.py b/policyengine_api/services/policy_mirroring.py index 50bb585ef..b71dc0d35 100644 --- a/policyengine_api/services/policy_mirroring.py +++ b/policyengine_api/services/policy_mirroring.py @@ -5,6 +5,7 @@ import contextlib import time from dataclasses import dataclass +from enum import StrEnum from typing import TYPE_CHECKING, Protocol from sqlalchemy.exc import SQLAlchemyError @@ -29,6 +30,7 @@ from policyengine_api.services.v2.policies.types import ( LegacyPolicyPersistenceResult, LegacyPolicySnapshot, + LegacyPolicyTranslationReason, ) @@ -59,31 +61,142 @@ def _default_mirror_factory() -> LegacyPolicyMirror: return V2PolicyService(PolicyDatabaseSession(get_v2_session_factory())) -def _failure_category(error: Exception) -> str: - if isinstance(error, V2ConfigurationError): - return "configuration" - if isinstance( - error, - ( - MetadataCatalogUnavailableError, - MetadataCatalogVersionNotFoundError, - PolicyCatalogValidationError, - LegacyPolicyTranslationError, +class PolicyCopyFailureCategory(StrEnum): + """The broad kinds of failed copy, as a copy record's ``failure_category``.""" + + CONFIGURATION = "configuration" + CATALOG_OR_TRANSLATION = "catalog_or_translation" + INTEGRITY = "integrity" + DATABASE = "database" + UNEXPECTED = "unexpected" + # The route logs these when no copy could be attempted. + MISSING_MIRROR_SNAPSHOT = "missing_mirror_snapshot" + MISSING_MIRROR_REVISION = "missing_mirror_revision" + + +class PolicyCopyFailureReason(StrEnum): + """Fixed reason codes that failed policy and saved-policy copies log. + + A translation failure logs its LegacyPolicyTranslationReason instead, such + as ``missing_parameters``. The two sets together are every value of a copy + record's ``failure_reason``. + """ + + CONFIGURATION = "configuration" + CATALOG_QUERY_FAILED = "catalog_query_failed" + CATALOG_ABSENT = "catalog_absent" + CATALOG_VERSION_NOT_FOUND = "catalog_version_not_found" + CATALOG_MISMATCH = "catalog_mismatch" + MAPPING_CONFLICT = "mapping_conflict" + CONTENT_HASH_COLLISION = "content_hash_collision" + STORED_POLICY_INTEGRITY = "stored_policy_integrity" + SAVED_POLICY_INTEGRITY = "saved_policy_integrity" + MISSING_REFORM_POLICY = "missing_reform_policy" + MISSING_MIRROR_SNAPSHOT = "missing_mirror_snapshot" + MISSING_MIRROR_REVISION = "missing_mirror_revision" + DATABASE_ERROR = "database_error" + UNEXPECTED = "unexpected" + + +@dataclass(frozen=True) +class PolicyCopyFailure: + """How a failed copy is logged: a broad category and a fixed reason.""" + + category: PolicyCopyFailureCategory + reason: PolicyCopyFailureReason | LegacyPolicyTranslationReason + + +# What the route logs when it commits a v1 write but cannot attempt a copy. +MISSING_MIRROR_SNAPSHOT_FAILURE = PolicyCopyFailure( + PolicyCopyFailureCategory.MISSING_MIRROR_SNAPSHOT, + PolicyCopyFailureReason.MISSING_MIRROR_SNAPSHOT, +) +MISSING_MIRROR_REVISION_FAILURE = PolicyCopyFailure( + PolicyCopyFailureCategory.MISSING_MIRROR_REVISION, + PolicyCopyFailureReason.MISSING_MIRROR_REVISION, +) +UNEXPECTED_FAILURE = PolicyCopyFailure( + PolicyCopyFailureCategory.UNEXPECTED, PolicyCopyFailureReason.UNEXPECTED +) + + +# Checked in order, so the hash collision is matched before its integrity base +# class. +_FAILURES_BY_ERROR_TYPE: tuple[tuple[type[Exception], PolicyCopyFailure], ...] = ( + ( + V2ConfigurationError, + PolicyCopyFailure( + PolicyCopyFailureCategory.CONFIGURATION, + PolicyCopyFailureReason.CONFIGURATION, + ), + ), + ( + MetadataCatalogVersionNotFoundError, + PolicyCopyFailure( + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, + PolicyCopyFailureReason.CATALOG_VERSION_NOT_FOUND, + ), + ), + ( + PolicyCatalogValidationError, + PolicyCopyFailure( + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, + PolicyCopyFailureReason.CATALOG_MISMATCH, ), - ): - return "catalog_or_translation" - if isinstance( - error, - ( - LegacyPolicyMappingIntegrityError, - PolicyContentHashCollisionError, - PolicyCreationIntegrityError, + ), + ( + LegacyPolicyMappingIntegrityError, + PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.MAPPING_CONFLICT, ), - ): - return "integrity" - if isinstance(error, SQLAlchemyError): - return "database" - return "unexpected" + ), + ( + PolicyContentHashCollisionError, + PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.CONTENT_HASH_COLLISION, + ), + ), + ( + PolicyCreationIntegrityError, + PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.STORED_POLICY_INTEGRITY, + ), + ), + ( + SQLAlchemyError, + PolicyCopyFailure( + PolicyCopyFailureCategory.DATABASE, PolicyCopyFailureReason.DATABASE_ERROR + ), + ), +) + + +def classify_policy_copy_failure(error: Exception) -> PolicyCopyFailure: + """Return how a failed policy copy logs this error. + + Both fields come only from the error's type, and for a translation error + its fixed reason, so a log record never holds policy content or error text. + """ + if isinstance(error, LegacyPolicyTranslationError): + return PolicyCopyFailure( + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, error.reason + ) + if isinstance(error, MetadataCatalogUnavailableError): + reason = ( + PolicyCopyFailureReason.CATALOG_QUERY_FAILED + if isinstance(error.__cause__, SQLAlchemyError) + else PolicyCopyFailureReason.CATALOG_ABSENT + ) + return PolicyCopyFailure( + PolicyCopyFailureCategory.CATALOG_OR_TRANSLATION, reason + ) + for error_type, failure in _FAILURES_BY_ERROR_TYPE: + if isinstance(error, error_type): + return failure + return UNEXPECTED_FAILURE @dataclass(frozen=True) @@ -93,7 +206,7 @@ class _MirrorOutcome: outcome: str actual_write_sources: list[str] destination_policy_id: object | None = None - failure_category: str | None = None + failure: PolicyCopyFailure | None = None policy_created: bool | None = None mapping_created: bool | None = None @@ -105,6 +218,7 @@ def _log_mirror_event( mirror_outcome: _MirrorOutcome, ) -> None: destination_policy_id = mirror_outcome.destination_policy_id + failure = mirror_outcome.failure # Metric logging must not change the mirror outcome. with contextlib.suppress(Exception): logger.log_struct( @@ -123,7 +237,8 @@ def _log_mirror_event( else None ), "outcome": mirror_outcome.outcome, - "failure_category": mirror_outcome.failure_category, + "failure_category": failure.category.value if failure else None, + "failure_reason": failure.reason.value if failure else None, "policy_created": mirror_outcome.policy_created, "mapping_created": mirror_outcome.mapping_created, "duration_ms": round((time.perf_counter() - started_at) * 1000, 3), @@ -154,7 +269,7 @@ def mirror_policy_after_commit( mirror_outcome=_MirrorOutcome( outcome="error", actual_write_sources=["cloud_sql"], - failure_category=_failure_category(error), + failure=classify_policy_copy_failure(error), ), ) msg = "The committed v1 policy could not be mirrored to v2" diff --git a/policyengine_api/services/user_policy_mirroring.py b/policyengine_api/services/user_policy_mirroring.py index e3f14542b..d72d7f7f5 100644 --- a/policyengine_api/services/user_policy_mirroring.py +++ b/policyengine_api/services/user_policy_mirroring.py @@ -7,26 +7,18 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Protocol -from sqlalchemy.exc import SQLAlchemyError - -from policyengine_api.data.v2.catalog.catalog_selection import ( - MetadataCatalogUnavailableError, - MetadataCatalogVersionNotFoundError, -) -from policyengine_api.data.v2.settings import V2ConfigurationError from policyengine_api.gcp_logging import logger +from policyengine_api.services.policy_mirroring import ( + PolicyCopyFailure, + PolicyCopyFailureCategory, + PolicyCopyFailureReason, + classify_policy_copy_failure, +) from policyengine_api.services.policy_service import PolicyService from policyengine_api.services.user_policy_service import ( PendingUserPolicyMirrorEvent, UserPolicyService, ) -from policyengine_api.services.v2.policies.validators import ( - LegacyPolicyMappingIntegrityError, - LegacyPolicyTranslationError, - PolicyCatalogValidationError, - PolicyContentHashCollisionError, - PolicyCreationIntegrityError, -) from policyengine_api.services.v2.user_policies.validators import ( LegacyUserPolicyIntegrityError, ) @@ -60,6 +52,10 @@ class UserPolicyMirrorUnavailableError(RuntimeError): """Raised when a committed saved policy could not be mirrored immediately.""" +class ReformPolicyUnavailableError(UserPolicyMirrorUnavailableError): + """Raised when a saved policy's v1 reform policy cannot be loaded to copy it.""" + + def _default_mirror_factory() -> LegacyUserPolicyMirror: from policyengine_api.data.v2.database import ( get_v2_session_factory, @@ -72,32 +68,18 @@ def _default_mirror_factory() -> LegacyUserPolicyMirror: return V2UserPolicyService(UserPolicyDatabaseSession(get_v2_session_factory())) -def _failure_category(error: Exception) -> str: - if isinstance(error, V2ConfigurationError): - return "configuration" - if isinstance( - error, - ( - MetadataCatalogUnavailableError, - MetadataCatalogVersionNotFoundError, - PolicyCatalogValidationError, - LegacyPolicyTranslationError, - ), - ): - return "catalog_or_translation" - if isinstance( - error, - ( - LegacyPolicyMappingIntegrityError, - LegacyUserPolicyIntegrityError, - PolicyContentHashCollisionError, - PolicyCreationIntegrityError, - ), - ): - return "integrity" - if isinstance(error, SQLAlchemyError): - return "database" - return "unexpected" +def _classify_failure(error: Exception) -> PolicyCopyFailure: + if isinstance(error, LegacyUserPolicyIntegrityError): + return PolicyCopyFailure( + PolicyCopyFailureCategory.INTEGRITY, + PolicyCopyFailureReason.SAVED_POLICY_INTEGRITY, + ) + if isinstance(error, ReformPolicyUnavailableError): + return PolicyCopyFailure( + PolicyCopyFailureCategory.UNEXPECTED, + PolicyCopyFailureReason.MISSING_REFORM_POLICY, + ) + return classify_policy_copy_failure(error) @dataclass(frozen=True) @@ -134,7 +116,7 @@ class _MirrorOutcome: outcome: str actual_write_sources: list[str] result: LegacyUserPolicyPersistenceResult | None = None - failure_category: str | None = None + failure: PolicyCopyFailure | None = None def _log_mirror_operation( @@ -142,6 +124,7 @@ def _log_mirror_operation( mirror_outcome: _MirrorOutcome, ) -> None: result = mirror_outcome.result + failure = mirror_outcome.failure # Metric logging must not change the mirror outcome. with contextlib.suppress(Exception): logger.log_struct( @@ -163,7 +146,8 @@ def _log_mirror_operation( str(result.policy_id) if result is not None else None ), "outcome": mirror_outcome.outcome, - "failure_category": mirror_outcome.failure_category, + "failure_category": failure.category.value if failure else None, + "failure_reason": failure.reason.value if failure else None, "association_created": ( result.association_created if result is not None else None ), @@ -231,7 +215,7 @@ def mirror_user_policy_after_commit( _MirrorOutcome( outcome="error", actual_write_sources=["cloud_sql"], - failure_category=_failure_category(error), + failure=_classify_failure(error), ), ) msg = "The committed v1 saved policy could not be mirrored to v2" @@ -292,7 +276,7 @@ def process( ) if reform_snapshot is None: msg = "The saved policy references an unavailable reform policy" - raise UserPolicyMirrorUnavailableError(msg) + raise ReformPolicyUnavailableError(msg) result = _run_user_policy_mirror( event.snapshot, reform_snapshot, @@ -366,7 +350,7 @@ def after_processed_commit( ["cloud_sql", "supabase"] if result is not None else ["cloud_sql"] ), result=result, - failure_category=_failure_category(failure), + failure=_classify_failure(failure), ), ) msg = "The committed v1 saved policy event could not be mirrored to v2" diff --git a/policyengine_api/services/v2/policies/services.py b/policyengine_api/services/v2/policies/services.py index 5c21c124d..eb8250fd1 100644 --- a/policyengine_api/services/v2/policies/services.py +++ b/policyengine_api/services/v2/policies/services.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING -from policyengine_api.constants import COUNTRY_PACKAGE_VERSIONS, POLICYENGINE_VERSION +from policyengine_api.constants import POLICYENGINE_VERSION from policyengine_api.services.v2.policies.database_connectors.creates import ( create_legacy_policy_mapping, create_parameter_values, @@ -53,7 +53,7 @@ ) if TYPE_CHECKING: - from collections.abc import Callable, Mapping + from collections.abc import Callable from uuid import UUID from sqlmodel import Session @@ -190,13 +190,15 @@ def mirror_legacy_policy_in_session( snapshot: LegacyPolicySnapshot, *, running_policyengine_version: str = POLICYENGINE_VERSION, - country_package_versions: Mapping[str, str] = COUNTRY_PACKAGE_VERSIONS, ) -> LegacyPolicyPersistenceResult: """Copy one committed v1 policy into v2 and map its v1 id to the result. - The policy is translated against the running catalog. Replaying the same v1 - policy is idempotent. Raise LegacyPolicyMappingIntegrityError if the v1 id is - already mapped with a different source hash or to a different v2 policy. + The policy is translated against the running catalog, whichever country + package version it was saved under. A v1 policy that already has a v2 copy + returns that copy without being translated again, so writing it again after + a PolicyEngine.py or country package upgrade still succeeds. Raise + LegacyPolicyMappingIntegrityError if the v1 id is already mapped with a + different source hash. """ existing = read_legacy_policy_mapping( session, @@ -205,9 +207,16 @@ def mirror_legacy_policy_in_session( lock=True, ) if existing is not None: + # v1 policies never change, so the same source hash means the mapped + # v2 policy is still the copy, whichever catalog is running now. verify_legacy_policy_mapping( existing, source_policy_hash=snapshot.source_policy_hash ) + return LegacyPolicyPersistenceResult( + policy_id=existing.policy_id, + policy_created=False, + mapping_created=False, + ) selected = read_policy_catalog( session, @@ -227,21 +236,8 @@ def mirror_legacy_policy_in_session( snapshot, selected=selected, parameters=parameters, - country_package_versions=country_package_versions, ) policy_result = create_resolved_policy(session, policy_input) - if existing is not None: - verify_legacy_policy_mapping( - existing, - source_policy_hash=snapshot.source_policy_hash, - expected_policy_id=policy_result.policy_id, - ) - return LegacyPolicyPersistenceResult( - policy_id=existing.policy_id, - policy_created=False, - mapping_created=False, - ) - mapping_id = create_legacy_policy_mapping( session, country_id=snapshot.country_id, @@ -264,10 +260,14 @@ def mirror_legacy_policy_in_session( if concurrent is None: msg = "legacy policy mapping conflict did not resolve to a stored row" raise LegacyPolicyMappingIntegrityError(msg) + # A concurrent copy mapped the policy first. As with a mapping found at the + # start, the same source hash makes its v2 policy the copy, even when it was + # translated with another catalog version than this one. If this copy's + # translation differed, the v2 policy it inserted commits without a mapping, + # and the result's policy_created is false because it reports the mapped + # policy, not that one. verify_legacy_policy_mapping( - concurrent, - source_policy_hash=snapshot.source_policy_hash, - expected_policy_id=policy_result.policy_id, + concurrent, source_policy_hash=snapshot.source_policy_hash ) return LegacyPolicyPersistenceResult( policy_id=concurrent.policy_id, diff --git a/policyengine_api/services/v2/policies/transformations.py b/policyengine_api/services/v2/policies/transformations.py index 5fceabf51..8b20a6c74 100644 --- a/policyengine_api/services/v2/policies/transformations.py +++ b/policyengine_api/services/v2/policies/transformations.py @@ -15,6 +15,7 @@ from policyengine_api.services.v2.policies.types import ( CanonicalPolicyContent, LegacyPolicySnapshot, + LegacyPolicyTranslationReason, PolicyCreationInput, PolicyPage, PolicyParameterValueInput, @@ -23,7 +24,9 @@ ResolvedPolicyCreationInput, ) from policyengine_api.services.v2.policies.validators import ( + MAXIMUM_POLICY_PARAMETER_VALUES, LegacyPolicyTranslationError, + parameters_missing_from_model, validate_policy_catalog, ) @@ -221,7 +224,9 @@ def _utc_midnight(value: str) -> datetime: parsed = date.fromisoformat(value) except ValueError as error: msg = f"legacy period {value!r} is invalid" - raise LegacyPolicyTranslationError(msg) from error + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.INVALID_PERIOD + ) from error return datetime.combine(parsed, time.min, tzinfo=UTC) @@ -231,14 +236,15 @@ def parse_legacy_period(value: str) -> tuple[datetime, datetime]: Accept a ``start.end`` ISO date range or a PolicyEngine period such as ``2024``. Raise LegacyPolicyTranslationError for anything else. """ + invalid_period = LegacyPolicyTranslationReason.INVALID_PERIOD if not value or value != value.strip(): msg = "legacy period must be non-empty" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError(msg, reason=invalid_period) if "." in value: parts = value.split(".") if len(parts) != LEGACY_DATE_RANGE_PART_COUNT: msg = f"legacy period {value!r} must contain one date range" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError(msg, reason=invalid_period) start_date, end_date = (_utc_midnight(item) for item in parts) else: try: @@ -247,10 +253,10 @@ def parse_legacy_period(value: str) -> tuple[datetime, datetime]: end_date = _utc_midnight(str(parsed.stop)) except (TypeError, ValueError) as error: msg = f"legacy period {value!r} is invalid" - raise LegacyPolicyTranslationError(msg) from error + raise LegacyPolicyTranslationError(msg, reason=invalid_period) from error if end_date < start_date: msg = f"legacy period {value!r} ends before it starts" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError(msg, reason=invalid_period) return start_date, end_date @@ -259,39 +265,49 @@ def translate_legacy_policy( *, selected: SelectedCatalog, parameters: Mapping[str, Parameter], - country_package_versions: Mapping[str, str], ) -> ResolvedPolicyCreationInput: """Translate a v1 policy into policy content bound to the running catalog. - Raise LegacyPolicyTranslationError if the v1 policy was saved under another - country package version, names a parameter the catalog lacks, or has - invalid or conflicting periods. + The country package version the v1 policy was saved under is not compared: + the Public API simulates an old policy with the running package, so the copy + binds it to the running catalog the same way. + + ``parameters`` holds the running catalog's parameters with the policy's + parameter paths, keyed by path. Raise LegacyPolicyTranslationError, with a + fixed reason, if the v1 policy sets a parameter the catalog lacks, sets no + value for a parameter, or has invalid or conflicting periods. """ - expected_api_version = country_package_versions.get(snapshot.country_id) - if expected_api_version is None or snapshot.api_version != expected_api_version: - msg = "legacy policy api_version does not match the running country package" - raise LegacyPolicyTranslationError(msg) policy_json = snapshot.policy_json if not isinstance(policy_json, dict): msg = "legacy policy_json must be an object" raise TypeError(msg) - parameter_names = set(policy_json) - if set(parameters) != parameter_names: + if parameters_missing_from_model(policy_json, parameters): msg = "every legacy parameter path must exist in the running catalog" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.MISSING_PARAMETERS + ) parameter_values: list[PolicyParameterValueInput] = [] - for parameter_name in sorted(parameter_names): + for parameter_name in sorted(policy_json): period_values = policy_json[parameter_name] if type(period_values) is not dict: msg = ( f"legacy parameter {parameter_name!r} must contain period/value entries" ) - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.INVALID_PARAMETER_VALUES + ) + if not period_values: + msg = f"legacy parameter {parameter_name!r} sets no values" + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.EMPTY_PARAMETER_VALUES + ) for period_name, value in sorted(period_values.items()): if type(period_name) is not str: msg = f"legacy parameter {parameter_name!r} has a non-string period" - raise LegacyPolicyTranslationError(msg) + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.INVALID_PERIOD + ) start_date, end_date = parse_legacy_period(period_name) parameter_values.append( PolicyParameterValueInput( @@ -301,6 +317,11 @@ def translate_legacy_policy( end_date=end_date, ) ) + if len(parameter_values) > MAXIMUM_POLICY_PARAMETER_VALUES: + msg = "legacy policy has more values than a v2 policy may hold" + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.TOO_MANY_VALUES + ) try: policy_input = PolicyCreationInput( country_id=snapshot.country_id, @@ -308,8 +329,12 @@ def translate_legacy_policy( parameter_values=parameter_values, ) except ValueError as error: + # The count is checked above, so what remains is two values for one + # parameter with the same start date. msg = "legacy parameter periods or values conflict" - raise LegacyPolicyTranslationError(msg) from error + raise LegacyPolicyTranslationError( + msg, reason=LegacyPolicyTranslationReason.CONFLICTING_VALUES + ) from error return validate_policy_catalog( policy_input, selected=selected, diff --git a/policyengine_api/services/v2/policies/types.py b/policyengine_api/services/v2/policies/types.py index 1c6dc15e2..3783d5e82 100644 --- a/policyengine_api/services/v2/policies/types.py +++ b/policyengine_api/services/v2/policies/types.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from datetime import datetime +from enum import StrEnum from typing import Annotated, Any from uuid import UUID @@ -26,6 +27,19 @@ ) StrictJsonValue = Annotated[JsonValue, BeforeValidator(require_json_value)] + + +class LegacyPolicyTranslationReason(StrEnum): + """Why committed v1 content cannot be translated, as copy failure logs name it.""" + + MISSING_PARAMETERS = "missing_parameters" + INVALID_PARAMETER_VALUES = "invalid_parameter_values" + EMPTY_PARAMETER_VALUES = "empty_parameter_values" + INVALID_PERIOD = "invalid_period" + CONFLICTING_VALUES = "conflicting_values" + TOO_MANY_VALUES = "too_many_values" + + UtcDateTime = Annotated[datetime, AfterValidator(normalize_utc)] diff --git a/policyengine_api/services/v2/policies/validators.py b/policyengine_api/services/v2/policies/validators.py index 4302d07da..ed7a1bbe2 100644 --- a/policyengine_api/services/v2/policies/validators.py +++ b/policyengine_api/services/v2/policies/validators.py @@ -7,11 +7,13 @@ from typing import TYPE_CHECKING, Any if TYPE_CHECKING: + from collections.abc import Container, Iterable from uuid import UUID from policyengine_api.data.v2.catalog.catalog_selection import SelectedCatalog from policyengine_api.data.v2.models import LegacyPolicyMapping from policyengine_api.services.v2.policies.types import ( + LegacyPolicyTranslationReason, PolicyCreationInput, ResolvedPolicyCreationInput, ) @@ -42,7 +44,36 @@ class LegacyPolicyMappingIntegrityError(RuntimeError): class LegacyPolicyTranslationError(ValueError): - """Raised when committed v1 content cannot be interpreted exactly.""" + """Raised when committed v1 content cannot be interpreted exactly. + + ``reason`` is a fixed code that copy failure logs record; the message is + never logged. + """ + + def __init__(self, message: str, *, reason: LegacyPolicyTranslationReason) -> None: + """Keep the message for debugging and the fixed reason for logs.""" + super().__init__(message) + self.reason = reason + + +def parameters_missing_from_model( + policy_parameter_names: Iterable[str], + model_parameter_names: Container[str], +) -> tuple[str, ...]: + """Return the parameter paths a policy sets that the model does not have. + + Each absent path appears once, in sorted order; an empty tuple means every + path exists. The v2 copy of a v1 policy is refused when it is not empty. + """ + return tuple( + sorted( + { + name + for name in policy_parameter_names + if name not in model_parameter_names + } + ) + ) def require_json_value( @@ -137,12 +168,12 @@ def verify_legacy_policy_mapping( mapping: LegacyPolicyMapping, *, source_policy_hash: str, - expected_policy_id: UUID | None = None, ) -> None: - """Validate an existing mapping without performing SQL.""" + """Require an existing mapping to come from the same v1 content, without SQL. + + v1 policies never change, so a mapping with the same source hash is the + copy whichever catalog version translated it. + """ if mapping.source_policy_hash != source_policy_hash: msg = "legacy policy identity was presented with a different source hash" raise LegacyPolicyMappingIntegrityError(msg) - if expected_policy_id is not None and mapping.policy_id != expected_policy_id: - msg = "legacy policy mapping does not match translated immutable content" - raise LegacyPolicyMappingIntegrityError(msg) diff --git a/tests/integration/test_mysql_policy_dual_write.py b/tests/integration/test_mysql_policy_dual_write.py index e3a778192..b7e678d05 100644 --- a/tests/integration/test_mysql_policy_dual_write.py +++ b/tests/integration/test_mysql_policy_dual_write.py @@ -48,9 +48,6 @@ 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 @@ -379,7 +376,7 @@ def create_native(value): assert postgres_destination.counts() == (2, 2, 1) -def test_historical_pre_storage_mapping_is_rejected_without_repair( +def test_historical_pre_storage_mapping_is_reused_without_repair( mysql_source, postgres_destination, record_property ) -> None: creation = mysql_source.create() @@ -397,13 +394,13 @@ def test_historical_pre_storage_mapping_is_rejected_without_repair( 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__) + # A v1 policy that already has a copy is not translated again: the retry + # returns the existing copy and neither repairs nor duplicates it. + reused = postgres_destination.mirror(retry) + assert reused.policy_id == original.policy_id + assert (reused.policy_created, reused.mapping_created) == (False, False) 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( diff --git a/tests/integration/test_v2_policy_persistence.py b/tests/integration/test_v2_policy_persistence.py index b53538ddd..7597f5c3d 100644 --- a/tests/integration/test_v2_policy_persistence.py +++ b/tests/integration/test_v2_policy_persistence.py @@ -253,13 +253,11 @@ def test_legacy_policy_mapping_is_many_to_one_and_retry_safe() -> None: session, first, running_policyengine_version=version.version, - country_package_versions={"us": "1.0.0"}, ) second_result = mirror_legacy_policy_in_session( session, second, running_policyengine_version=version.version, - country_package_versions={"us": "1.0.0"}, ) with Session(engine) as session, session.begin(): @@ -267,7 +265,6 @@ def test_legacy_policy_mapping_is_many_to_one_and_retry_safe() -> None: session, first, running_policyengine_version="5.2.0", - country_package_versions={"us": "1.0.0"}, ) assert first_result.policy_id == second_result.policy_id == retry.policy_id @@ -310,7 +307,6 @@ def test_changed_hash_for_one_legacy_identity_rolls_back_without_mutation() -> N session, snapshot, running_policyengine_version=version.version, - country_package_versions={"us": "1.0.0"}, ) changed = snapshot.model_copy( @@ -328,7 +324,6 @@ def test_changed_hash_for_one_legacy_identity_rolls_back_without_mutation() -> N session, changed, running_policyengine_version="5.2.0", - country_package_versions={"us": "1.0.0"}, ) with Session(engine) as session: diff --git a/tests/unit/routes/test_policy_dual_write_routes.py b/tests/unit/routes/test_policy_dual_write_routes.py index d644a1d29..87f5a3d52 100644 --- a/tests/unit/routes/test_policy_dual_write_routes.py +++ b/tests/unit/routes/test_policy_dual_write_routes.py @@ -13,7 +13,10 @@ from policyengine_api.routes.policy_routes import policy_bp from policyengine_api.services.policy_mirroring import PolicyMirrorUnavailableError from policyengine_api.services.policy_service import PolicySetResult -from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.v2.policies.types import ( + LegacyPolicySnapshot, + LegacyPolicyTranslationReason, +) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyTranslationError, ) @@ -224,7 +227,8 @@ def test_destination_failure_returns_the_stored_policy_and_logs_an_error( monkeypatch.setenv("DB_WRITE_POLICY", "dual_write") failing_destination = MagicMock() failing_destination.mirror_legacy_policy.side_effect = LegacyPolicyTranslationError( - "private parameter value" + "private parameter value", + reason=LegacyPolicyTranslationReason.MISSING_PARAMETERS, ) with ( patch( @@ -248,6 +252,7 @@ def test_destination_failure_returns_the_stored_policy_and_logs_an_error( assert logged_record["metric_name"] == "v1_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "catalog_or_translation" + assert logged_record["failure_reason"] == "missing_parameters" assert logged_record["legacy_policy_id"] == stored_policy_id assert "private parameter value" not in json.dumps(logged_record) @@ -281,6 +286,7 @@ def test_policy_without_a_mirror_snapshot_is_returned_and_logged( assert logged_record["metric_name"] == "v1_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "missing_mirror_snapshot" + assert logged_record["failure_reason"] == "missing_mirror_snapshot" assert logged_record["country_id"] == "us" assert logged_record["legacy_policy_id"] == 42 assert route_logger.log_struct.call_args.kwargs["severity"] == "ERROR" diff --git a/tests/unit/routes/test_user_policy_dual_write_routes.py b/tests/unit/routes/test_user_policy_dual_write_routes.py index 7233002b4..c36f57436 100644 --- a/tests/unit/routes/test_user_policy_dual_write_routes.py +++ b/tests/unit/routes/test_user_policy_dual_write_routes.py @@ -451,6 +451,7 @@ def test_unexpected_saved_policy_mirror_error_is_logged_and_the_record_returned( assert logged_record["metric_name"] == "v1_user_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "unexpected" + assert logged_record["failure_reason"] == "unexpected" assert logged_record["legacy_user_policy_id"] == 10 assert "secret" not in json.dumps(logged_record) assert route_logger.log_struct.call_args.kwargs["severity"] == "ERROR" @@ -480,6 +481,7 @@ def test_saved_policy_without_a_mirror_revision_is_returned_and_logged( assert logged_record["metric_name"] == "v1_user_policy_mirror_operations" assert logged_record["outcome"] == "error" assert logged_record["failure_category"] == "missing_mirror_revision" + assert logged_record["failure_reason"] == "missing_mirror_revision" assert logged_record["legacy_user_policy_id"] == 10 diff --git a/tests/unit/services/test_mirror_log_payloads.py b/tests/unit/services/test_mirror_log_payloads.py index 493b600de..426ba4169 100644 --- a/tests/unit/services/test_mirror_log_payloads.py +++ b/tests/unit/services/test_mirror_log_payloads.py @@ -204,6 +204,7 @@ def test_policy_mirror_success_logs_one_complete_record() -> None: "destination_policy_id": str(DESTINATION_ID), "outcome": "ok", "failure_category": None, + "failure_reason": None, "policy_created": False, "mapping_created": True, } @@ -235,6 +236,7 @@ def test_policy_mirror_failure_logs_one_complete_record() -> None: "destination_policy_id": None, "outcome": "error", "failure_category": "database", + "failure_reason": "database_error", "policy_created": None, "mapping_created": None, } @@ -285,6 +287,7 @@ def _user_policy_record(**changes: object) -> dict[str, object]: "destination_policy_id": str(DESTINATION_ID), "outcome": "ok", "failure_category": None, + "failure_reason": None, "association_created": False, "association_updated": True, "mapping_created": False, @@ -299,6 +302,7 @@ def _user_policy_record(**changes: object) -> dict[str, object]: destination_policy_id=None, outcome="error", failure_category="database", + failure_reason="database_error", association_created=None, association_updated=None, mapping_created=None, @@ -429,6 +433,7 @@ def test_pending_event_failure_logs_the_active_event_and_unwrapped_cause() -> No legacy_user_policy_id=11, outcome="error", failure_category="database", + failure_reason="database_error", ) assert list(payload) == list(expected) assert payload == expected diff --git a/tests/unit/services/test_policy_mirroring.py b/tests/unit/services/test_policy_mirroring.py index ad2bdaf3f..8459e0a0f 100644 --- a/tests/unit/services/test_policy_mirroring.py +++ b/tests/unit/services/test_policy_mirroring.py @@ -9,16 +9,29 @@ from sqlalchemy.exc import OperationalError from sqlalchemy.exc import TimeoutError as PoolTimeoutError +from policyengine_api.data.v2.catalog.catalog_selection import ( + MetadataCatalogUnavailableError, + MetadataCatalogVersionNotFoundError, +) +from policyengine_api.data.v2.settings import V2ConfigurationError from policyengine_api.services.policy_mirroring import ( + PolicyCopyFailureCategory, + PolicyCopyFailureReason, PolicyMirrorUnavailableError, + classify_policy_copy_failure, mirror_policy_after_commit, ) from policyengine_api.services.v2.policies.types import ( LegacyPolicyPersistenceResult, LegacyPolicySnapshot, + LegacyPolicyTranslationReason, ) from policyengine_api.services.v2.policies.validators import ( LegacyPolicyMappingIntegrityError, + LegacyPolicyTranslationError, + PolicyCatalogValidationError, + PolicyContentHashCollisionError, + PolicyCreationIntegrityError, ) POLICY_ID = UUID("00000000-0000-0000-0000-000000000010") @@ -63,6 +76,7 @@ def test_success_returns_destination_and_logs_metric_without_policy_content() -> "destination_policy_id": str(POLICY_ID), "outcome": "ok", "failure_category": None, + "failure_reason": None, "policy_created": True, "mapping_created": True, } @@ -123,3 +137,133 @@ def test_failure_is_secret_safe_and_records_cloud_sql_as_only_completed_write( assert payload["destination_policy_id"] is None assert "secret" not in repr(payload) assert "credential" not in repr(payload) + + +def _catalog_query_failure() -> MetadataCatalogUnavailableError: + error = MetadataCatalogUnavailableError("the v2 metadata catalog cannot be queried") + error.__cause__ = OperationalError("select secret", {}, Exception("credential")) + return error + + +def _translation_failure( + reason: LegacyPolicyTranslationReason, +) -> LegacyPolicyTranslationError: + return LegacyPolicyTranslationError("gov.private.secret", reason=reason) + + +@pytest.mark.parametrize( + ("error", "category", "reason"), + [ + pytest.param( + V2ConfigurationError("secret url"), + "configuration", + "configuration", + id="configuration", + ), + pytest.param( + _catalog_query_failure(), + "catalog_or_translation", + "catalog_query_failed", + id="catalog-query-failed", + ), + pytest.param( + MetadataCatalogUnavailableError("catalog absent secret"), + "catalog_or_translation", + "catalog_absent", + id="catalog-absent", + ), + pytest.param( + MetadataCatalogVersionNotFoundError("secret"), + "catalog_or_translation", + "catalog_version_not_found", + id="catalog-version-not-found", + ), + pytest.param( + PolicyCatalogValidationError("secret"), + "catalog_or_translation", + "catalog_mismatch", + id="catalog-mismatch", + ), + *( + pytest.param( + _translation_failure(reason), + "catalog_or_translation", + reason.value, + id=reason.value, + ) + for reason in LegacyPolicyTranslationReason + ), + pytest.param( + LegacyPolicyMappingIntegrityError("secret"), + "integrity", + "mapping_conflict", + id="mapping-conflict", + ), + pytest.param( + PolicyContentHashCollisionError("secret"), + "integrity", + "content_hash_collision", + id="content-hash-collision", + ), + pytest.param( + PolicyCreationIntegrityError("secret"), + "integrity", + "stored_policy_integrity", + id="stored-policy-integrity", + ), + pytest.param( + PoolTimeoutError("pool timeout secret"), + "database", + "database_error", + id="database", + ), + pytest.param( + RuntimeError("policy content secret"), + "unexpected", + "unexpected", + id="unexpected", + ), + ], +) +def test_each_failed_copy_logs_a_fixed_reason_with_its_category( + error: Exception, + category: str, + reason: str, +) -> None: + mirror = MagicMock() + mirror.mirror_legacy_policy.side_effect = error + + with ( + patch("policyengine_api.services.policy_mirroring.logger") as logger, + pytest.raises(PolicyMirrorUnavailableError), + ): + mirror_policy_after_commit(_snapshot(), mirror_factory=lambda: mirror) + + payload = logger.log_struct.call_args.args[0] + assert (payload["failure_category"], payload["failure_reason"]) == ( + category, + reason, + ) + assert type(payload["failure_reason"]) is str + assert "secret" not in repr(payload) + assert "credential" not in repr(payload) + + +@pytest.mark.parametrize( + "error", + [ + V2ConfigurationError("secret url"), + MetadataCatalogUnavailableError("catalog absent"), + _translation_failure(LegacyPolicyTranslationReason.MISSING_PARAMETERS), + LegacyPolicyMappingIntegrityError("secret"), + PoolTimeoutError("pool timeout"), + RuntimeError("policy content secret"), + ], +) +def test_failure_category_and_reason_come_from_fixed_sets(error: Exception) -> None: + failure = classify_policy_copy_failure(error) + + assert isinstance(failure.category, PolicyCopyFailureCategory) + assert isinstance( + failure.reason, (PolicyCopyFailureReason, LegacyPolicyTranslationReason) + ) diff --git a/tests/unit/services/test_user_policy_mirroring.py b/tests/unit/services/test_user_policy_mirroring.py index df8e7facf..de2e40cb4 100644 --- a/tests/unit/services/test_user_policy_mirroring.py +++ b/tests/unit/services/test_user_policy_mirroring.py @@ -18,8 +18,18 @@ mirror_pending_user_policy_events_after_commit, mirror_user_policy_after_commit, ) -from policyengine_api.services.user_policy_service import UserPolicyService -from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.user_policy_service import ( + PendingUserPolicyMirrorEvent, + UserPolicyService, +) +from policyengine_api.services.v2.policies.types import ( + LegacyPolicySnapshot, + LegacyPolicyTranslationReason, +) +from policyengine_api.services.v2.policies.validators import ( + LegacyPolicyMappingIntegrityError, + LegacyPolicyTranslationError, +) from policyengine_api.services.v2.user_policies.types import ( LegacyUserPolicyPersistenceResult, LegacyUserPolicySnapshot, @@ -131,20 +141,39 @@ def test_logging_failure_does_not_change_successful_mirror_result() -> None: @pytest.mark.parametrize( - ("error", "category"), + ("error", "category", "reason"), [ ( OperationalError("statement secret", {}, Exception("credential")), "database", + "database_error", + ), + (PoolTimeoutError("pool timeout"), "database", "database_error"), + ( + LegacyUserPolicyIntegrityError("mapping conflict"), + "integrity", + "saved_policy_integrity", + ), + ( + LegacyPolicyMappingIntegrityError("reform mapping conflict"), + "integrity", + "mapping_conflict", ), - (PoolTimeoutError("pool timeout"), "database"), - (LegacyUserPolicyIntegrityError("mapping conflict"), "integrity"), - (RuntimeError("caller data secret"), "unexpected"), + ( + LegacyPolicyTranslationError( + "gov.private.secret", + reason=LegacyPolicyTranslationReason.MISSING_PARAMETERS, + ), + "catalog_or_translation", + "missing_parameters", + ), + (RuntimeError("caller data secret"), "unexpected", "unexpected"), ], ) def test_failure_is_secret_safe_and_reports_only_cloud_sql_completed( error: Exception, category: str, + reason: str, ) -> None: mirror = MagicMock() mirror.mirror_legacy_user_policy.side_effect = error @@ -163,6 +192,7 @@ def test_failure_is_secret_safe_and_reports_only_cloud_sql_completed( payload = logger.log_struct.call_args.args[0] assert payload["actual_write_sources"] == ["cloud_sql"] assert payload["failure_category"] == category + assert payload["failure_reason"] == reason assert payload["destination_association_id"] is None assert "secret" not in repr(payload) assert "credential" not in repr(payload) @@ -268,3 +298,40 @@ def test_event_preparation_failure_logs_error_and_retains_the_event( with orm_session_factory() as session: retained_event = session.scalar(select(UserPolicyMirrorEvent)) assert retained_event.processed_at is None + + +def test_saved_policy_whose_reform_is_missing_logs_its_own_reason() -> None: + event_service = MagicMock() + + def process(_country_id, _legacy_user_policy_id, **callbacks): + return callbacks["processor"]( + PendingUserPolicyMirrorEvent( + event_id=1, + source_revision=1, + event_type="create", + snapshot=_saved(), + changed_fields=frozenset(), + source_fingerprint_sha256="f" * 64, + ) + ) + + event_service.process_pending_mirror_events.side_effect = process + + with ( + patch("policyengine_api.services.user_policy_mirroring.logger") as logger, + pytest.raises(UserPolicyMirrorUnavailableError), + ): + mirror_pending_user_policy_events_after_commit( + "us", + 10, + through_revision=1, + collaborators=UserPolicyMirrorCollaborators( + event_service=event_service, + reform_snapshot_loader=lambda _country_id, _policy_id: None, + mirror_factory=MagicMock(), + ), + ) + + payload = logger.log_struct.call_args.args[0] + assert payload["failure_category"] == "unexpected" + assert payload["failure_reason"] == "missing_reform_policy" diff --git a/tests/unit/v2/test_policy_legacy_copy.py b/tests/unit/v2/test_policy_legacy_copy.py new file mode 100644 index 000000000..e3ea73d7b --- /dev/null +++ b/tests/unit/v2/test_policy_legacy_copy.py @@ -0,0 +1,352 @@ +"""Copy committed v1 policies into v2 with the running catalog. + +The catalog and mapping reads run against SQLite. The two inserts use +PostgreSQL's ON CONFLICT ON CONSTRAINT, so the tests record them instead. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from uuid import UUID + +import pytest +from sqlmodel import Session, create_engine + +from policyengine_api.data.v2.models import ( + V2_METADATA, + LegacyPolicyMapping, + Parameter, + TaxBenefitModel, + TaxBenefitModelVersion, +) +from policyengine_api.services.v2.policies import services as policy_services +from policyengine_api.services.v2.policies.types import ( + LegacyPolicyPersistenceResult, + LegacyPolicySnapshot, + PolicyCreationResult, + ResolvedPolicyCreationInput, +) +from policyengine_api.services.v2.policies.validators import ( + LegacyPolicyMappingIntegrityError, +) + +RUNNING_POLICYENGINE_VERSION = "6.7.5" +COPIED_POLICY_ID = UUID("00000000-0000-0000-0000-000000000042") +EARLIER_COPY_POLICY_ID = UUID("00000000-0000-0000-0000-000000000041") +MAPPING_ID = UUID("00000000-0000-0000-0000-000000000043") + + +@dataclass +class RecordedCopyWrites: + """The policies and mappings a copy would have inserted, in order.""" + + created_policies: list[ResolvedPolicyCreationInput] = field(default_factory=list) + created_mappings: list[dict[str, object]] = field(default_factory=list) + + +@pytest.fixture +def recorded_writes(monkeypatch) -> RecordedCopyWrites: + writes = RecordedCopyWrites() + + def record_policy(_session, policy_input): + writes.created_policies.append(policy_input) + return PolicyCreationResult(policy_id=COPIED_POLICY_ID, created=True) + + def record_mapping(_session, **mapping_fields): + writes.created_mappings.append(mapping_fields) + return MAPPING_ID + + monkeypatch.setattr(policy_services, "create_resolved_policy", record_policy) + monkeypatch.setattr(policy_services, "create_legacy_policy_mapping", record_mapping) + return writes + + +@pytest.fixture +def catalog_session(): + engine = create_engine("sqlite://") + V2_METADATA.create_all(engine) + session = Session(engine) + try: + yield session + finally: + session.close() + engine.dispose() + + +def _add_running_catalog( + session: Session, *, country_id: str, parameter_names: list[str] +) -> tuple[TaxBenefitModelVersion, dict[str, Parameter]]: + model = TaxBenefitModel(name=f"policyengine-{country_id}") + version = TaxBenefitModelVersion( + model=model, + version=RUNNING_POLICYENGINE_VERSION, + current_law_id=1, + metadata_time_periods=[2026], + ) + parameters = { + name: Parameter(name=name, tax_benefit_model_version=version) + for name in parameter_names + } + session.add(version) + session.add_all(parameters.values()) + session.commit() + return version, parameters + + +def _snapshot( + *, + country_id: str, + legacy_policy_id: int, + api_version: str, + policy_json: dict[str, object], +) -> LegacyPolicySnapshot: + return LegacyPolicySnapshot( + country_id=country_id, + legacy_policy_id=legacy_policy_id, + label=None, + api_version=api_version, + policy_json=policy_json, + source_policy_hash=f"legacy-hash-{legacy_policy_id}", + ) + + +def _copy(session: Session, snapshot: LegacyPolicySnapshot): + return policy_services.mirror_legacy_policy_in_session( + session, + snapshot, + running_policyengine_version=RUNNING_POLICYENGINE_VERSION, + ) + + +@dataclass(frozen=True) +class ReusedProductionPolicy: + """The shape of a v1 policy whose production copy failed.""" + + country_id: str + legacy_policy_id: int + saved_api_version: str + parameter_name: str + value: float + + +# The two production copies that failed after dual-write was turned on: each +# POST matched an existing v1 row saved under a country package release many +# versions older than the one the Public API ran (us 2.2.1, uk 2.130.1). +PRODUCTION_REUSED_POLICIES = [ + pytest.param( + ReusedProductionPolicy("us", 96567, "1.633.2", "gov.dol.minimum_wage", 9), + id="us-96567-saved-under-1.633.2", + ), + pytest.param( + ReusedProductionPolicy("uk", 97839, "2.89.2", "gov.boe.base_rate", 0.05), + id="uk-97839-saved-under-2.89.2", + ), +] + + +@pytest.mark.parametrize("reused_policy", PRODUCTION_REUSED_POLICIES) +def test_policy_saved_under_an_older_country_package_is_copied( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, + reused_policy: ReusedProductionPolicy, +) -> None: + country_id = reused_policy.country_id + parameter_name = reused_policy.parameter_name + version, parameters = _add_running_catalog( + catalog_session, country_id=country_id, parameter_names=[parameter_name] + ) + snapshot = _snapshot( + country_id=country_id, + legacy_policy_id=reused_policy.legacy_policy_id, + api_version=reused_policy.saved_api_version, + policy_json={parameter_name: {"2026-01-01.2100-12-31": reused_policy.value}}, + ) + + result = _copy(catalog_session, snapshot) + + assert result == LegacyPolicyPersistenceResult( + policy_id=COPIED_POLICY_ID, + policy_created=True, + mapping_created=True, + ) + [copied_policy] = recorded_writes.created_policies + assert copied_policy.country_id == country_id + assert copied_policy.tax_benefit_model_version_id == version.id + assert copied_policy.policyengine_version == RUNNING_POLICYENGINE_VERSION + [copied_value] = copied_policy.parameter_values + assert copied_value.parameter_id == parameters[parameter_name].id + assert copied_value.value == reused_policy.value + assert copied_value.start_date == datetime(2026, 1, 1, tzinfo=UTC) + assert copied_value.end_date == datetime(2100, 12, 31, tzinfo=UTC) + assert recorded_writes.created_mappings == [ + { + "country_id": country_id, + "legacy_policy_id": reused_policy.legacy_policy_id, + "source_policy_hash": snapshot.source_policy_hash, + "policy_id": COPIED_POLICY_ID, + } + ] + + +def test_policy_with_a_country_package_the_api_does_not_run_is_copied( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + # The copy binds content to the running catalog, as the Public API binds an + # old policy to the running country package when it simulates one. + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.example.rate"] + ) + snapshot = _snapshot( + country_id="us", + legacy_policy_id=501, + api_version="99.0.0", + policy_json={"gov.example.rate": {"2026": 0.2}}, + ) + + result = _copy(catalog_session, snapshot) + + assert result.mapping_created is True + assert len(recorded_writes.created_policies) == 1 + + +def _add_earlier_copy( + session: Session, snapshot: LegacyPolicySnapshot, *, source_policy_hash: str +) -> None: + session.add( + LegacyPolicyMapping( + country_id=snapshot.country_id, + legacy_policy_id=snapshot.legacy_policy_id, + policy_id=EARLIER_COPY_POLICY_ID, + source_policy_hash=source_policy_hash, + ) + ) + session.commit() + + +def _reused_snapshot() -> LegacyPolicySnapshot: + return _snapshot( + country_id="us", + legacy_policy_id=96567, + api_version="1.633.2", + policy_json={"gov.dol.minimum_wage": {"2026-01-01.2100-12-31": 9}}, + ) + + +def test_earlier_copy_is_reused_when_the_running_catalog_is_not_published( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + # After a PolicyEngine.py upgrade, the new release's catalog may not be + # published yet. Writing the same v1 policy again still succeeds. + snapshot = _reused_snapshot() + _add_earlier_copy( + catalog_session, snapshot, source_policy_hash=snapshot.source_policy_hash + ) + + result = _copy(catalog_session, snapshot) + + assert result == LegacyPolicyPersistenceResult( + policy_id=EARLIER_COPY_POLICY_ID, + policy_created=False, + mapping_created=False, + ) + assert recorded_writes == RecordedCopyWrites() + + +def test_earlier_copy_is_reused_when_the_running_catalog_lacks_its_parameter( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + # A later release renamed or removed the parameter. The v2 copy made + # before that keeps its own catalog version, so it is still the copy. + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.example.rate"] + ) + snapshot = _reused_snapshot() + _add_earlier_copy( + catalog_session, snapshot, source_policy_hash=snapshot.source_policy_hash + ) + + result = _copy(catalog_session, snapshot) + + assert result.policy_id == EARLIER_COPY_POLICY_ID + assert (result.policy_created, result.mapping_created) == (False, False) + assert recorded_writes == RecordedCopyWrites() + + +def test_earlier_copy_with_another_source_hash_is_an_integrity_error( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, +) -> None: + snapshot = _reused_snapshot() + _add_earlier_copy( + catalog_session, snapshot, source_policy_hash="different-legacy-hash" + ) + + with pytest.raises(LegacyPolicyMappingIntegrityError, match="source hash"): + _copy(catalog_session, snapshot) + + assert recorded_writes == RecordedCopyWrites() + + +def _lose_mapping_race(monkeypatch, *, winning_source_hash: str) -> None: + """Make the mapping insert find a row a concurrent copy committed first. + + The replacement returns None, as the insert does when its unique + constraint makes it a no-op. + """ + + def insert_after_concurrent_copy(session, **mapping_fields): + session.add( + LegacyPolicyMapping( + country_id=mapping_fields["country_id"], + legacy_policy_id=mapping_fields["legacy_policy_id"], + policy_id=EARLIER_COPY_POLICY_ID, + source_policy_hash=winning_source_hash, + ) + ) + session.flush() + + monkeypatch.setattr( + policy_services, "create_legacy_policy_mapping", insert_after_concurrent_copy + ) + + +def test_concurrent_copy_on_another_catalog_version_is_reused( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, + monkeypatch, +) -> None: + # Two first copies of the same v1 policy ran at once, for example on two + # revisions with different PolicyEngine.py versions, so each translated + # it to a different v2 policy. The mapping committed first is the copy. + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.dol.minimum_wage"] + ) + snapshot = _reused_snapshot() + _lose_mapping_race(monkeypatch, winning_source_hash=snapshot.source_policy_hash) + + result = _copy(catalog_session, snapshot) + + assert result == LegacyPolicyPersistenceResult( + policy_id=EARLIER_COPY_POLICY_ID, + policy_created=False, + mapping_created=False, + ) + assert len(recorded_writes.created_policies) == 1 + + +def test_concurrent_copy_with_another_source_hash_is_an_integrity_error( + catalog_session: Session, + recorded_writes: RecordedCopyWrites, + monkeypatch, +) -> None: + _add_running_catalog( + catalog_session, country_id="us", parameter_names=["gov.dol.minimum_wage"] + ) + _lose_mapping_race(monkeypatch, winning_source_hash="different-legacy-hash") + + with pytest.raises(LegacyPolicyMappingIntegrityError, match="source hash"): + _copy(catalog_session, _reused_snapshot()) diff --git a/tests/unit/v2/test_policy_legacy_translation.py b/tests/unit/v2/test_policy_legacy_translation.py index d5cf88753..07f69f3ef 100644 --- a/tests/unit/v2/test_policy_legacy_translation.py +++ b/tests/unit/v2/test_policy_legacy_translation.py @@ -30,9 +30,14 @@ parse_legacy_period, translate_legacy_policy, ) -from policyengine_api.services.v2.policies.types import LegacyPolicySnapshot +from policyengine_api.services.v2.policies.types import ( + LegacyPolicySnapshot, + LegacyPolicyTranslationReason, +) from policyengine_api.services.v2.policies.validators import ( + MAXIMUM_POLICY_PARAMETER_VALUES, LegacyPolicyTranslationError, + parameters_missing_from_model, ) from tests.fixtures.local_v1_database import create_test_v1_schema @@ -98,7 +103,6 @@ def _translate( snapshot, selected=selected, parameters=parameters, - country_package_versions={"us": "1.0.0"}, ) @@ -237,36 +241,113 @@ def unavailable_mirror(): @pytest.mark.parametrize( - "policy_json", + ("policy_json", "reason"), [ - {"gov.missing": {"2026": 1}}, - {"gov.example.rate": 1}, - {"gov.example.rate": {"not-a-period": 1}}, - { - "gov.example.rate": { - "2026": 1, - "2026-01-01.2026-12-31": 2, - } - }, + pytest.param( + {"gov.missing": {"2026": 1}}, + LegacyPolicyTranslationReason.MISSING_PARAMETERS, + id="unknown-path", + ), + pytest.param( + {"gov.example.rate": {"2026": 1}, "gov.removed": {"2026": 2}}, + LegacyPolicyTranslationReason.MISSING_PARAMETERS, + id="one-of-two-paths-unknown", + ), + pytest.param( + {"gov.example.rate": 1}, + LegacyPolicyTranslationReason.INVALID_PARAMETER_VALUES, + id="value-without-period", + ), + pytest.param( + {"gov.example.rate": {}}, + LegacyPolicyTranslationReason.EMPTY_PARAMETER_VALUES, + id="empty-period-object", + ), + pytest.param( + {"gov.example.rate": {"not-a-period": 1}}, + LegacyPolicyTranslationReason.INVALID_PERIOD, + id="unparseable-period", + ), + pytest.param( + {"gov.example.rate": {".": 1}}, + LegacyPolicyTranslationReason.INVALID_PERIOD, + id="dot-period", + ), + pytest.param( + { + "gov.example.rate": { + "2026": 1, + "2026-01-01.2026-12-31": 2, + } + }, + LegacyPolicyTranslationReason.CONFLICTING_VALUES, + id="two-values-same-start", + ), ], ) -def test_missing_paths_malformed_periods_and_conflicts_fail( +def test_untranslatable_policy_fails_with_a_fixed_reason( policy_json: dict[str, object], + reason: LegacyPolicyTranslationReason, ) -> None: engine, session, _model, _version, _first, _second = _session_and_catalog() try: - with pytest.raises(LegacyPolicyTranslationError): + with pytest.raises(LegacyPolicyTranslationError) as failure: _translate(session, _snapshot(policy_json=policy_json)) + assert failure.value.reason is reason finally: session.close() engine.dispose() -def test_country_package_version_must_match_running_release() -> None: +def test_policy_with_more_values_than_v2_accepts_fails_as_too_many_values() -> None: engine, session, _model, _version, _first, _second = _session_and_catalog() + # One more dated value than a v2 policy may hold, with no repeated dates. + too_many_periods = { + f"{2026 + offset}-01-01.{2026 + offset}-12-31": offset + for offset in range(MAXIMUM_POLICY_PARAMETER_VALUES + 1) + } + try: + with pytest.raises(LegacyPolicyTranslationError) as failure: + _translate( + session, + _snapshot(policy_json={"gov.example.rate": too_many_periods}), + ) + assert failure.value.reason is LegacyPolicyTranslationReason.TOO_MANY_VALUES + finally: + session.close() + engine.dispose() + + +def test_parameters_missing_from_model_lists_each_absent_name_once_in_order() -> None: + model_parameter_names = {"gov.example.rate", "gov.example.amount"} + + assert parameters_missing_from_model( + ["gov.removed.b", "gov.example.rate", "gov.removed.a", "gov.removed.b"], + model_parameter_names, + ) == ("gov.removed.a", "gov.removed.b") + assert parameters_missing_from_model([], model_parameter_names) == () + assert ( + parameters_missing_from_model( + {"gov.example.amount": {"2026": 1}}, model_parameter_names + ) + == () + ) + + +@pytest.mark.parametrize("saved_api_version", ["0.9.0", "1.633.2", "99.0.0"]) +def test_policy_saved_under_another_country_package_translates_to_running_catalog( + saved_api_version: str, +) -> None: + engine, session, _model, version, first, second = _session_and_catalog() try: - with pytest.raises(LegacyPolicyTranslationError, match="api_version"): - _translate(session, _snapshot(api_version="0.9.0")) + translated = _translate(session, _snapshot(api_version=saved_api_version)) + + assert translated.tax_benefit_model_version_id == version.id + assert {value.parameter_id for value in translated.parameter_values} == { + first.id, + second.id, + } + assert translated == _translate(session, _snapshot()) finally: session.close() engine.dispose() @@ -287,8 +368,9 @@ def test_snapshot_rejects_incomplete_or_non_json_committed_fields(changes) -> No def test_reverse_legacy_range_is_rejected() -> None: - with pytest.raises(LegacyPolicyTranslationError, match="ends before"): + with pytest.raises(LegacyPolicyTranslationError, match="ends before") as failure: parse_legacy_period("2026-12-31.2026-01-01") + assert failure.value.reason is LegacyPolicyTranslationReason.INVALID_PERIOD def test_unknown_policyengine_version_never_falls_back() -> None: