234 lines
9.2 KiB
Python
234 lines
9.2 KiB
Python
from flask import Flask
|
|
|
|
from services.pchome_mapping_backlog_service import (
|
|
build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package,
|
|
)
|
|
|
|
|
|
def _replay_package():
|
|
return {
|
|
"result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED",
|
|
"summary": {
|
|
"target_selector_count": 2,
|
|
"post_apply_readback_pass_count": 2,
|
|
"executor_receipt_hash_match_count": 1,
|
|
"missing_artifact_count": 0,
|
|
"writes_database_count": 0,
|
|
},
|
|
"missing_artifacts": [],
|
|
"receipt_replay": {"run_id": "run-test"},
|
|
}
|
|
|
|
|
|
def _drift_package(drift_count=0):
|
|
drift_items = []
|
|
if drift_count:
|
|
drift_items = [
|
|
{
|
|
"selector_id": "receipt-1",
|
|
"momo_icode": "MOMO-1",
|
|
"expected_pchome_id": "PCH-EXPECTED",
|
|
"actual_pchome_id": "PCH-DRIFT",
|
|
"expected_momo_name": "Expected product",
|
|
"actual_momo_name": "Expected product",
|
|
}
|
|
]
|
|
return {
|
|
"result": (
|
|
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED"
|
|
if drift_count
|
|
else "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED"
|
|
),
|
|
"summary": {
|
|
"target_selector_count": 2,
|
|
"post_apply_readback_pass_count": 2 - drift_count,
|
|
"drift_count": drift_count,
|
|
"drift_verified_count": 0 if drift_count else 1,
|
|
"drift_verifier_artifact_hash_match_count": 1,
|
|
"writes_database_count": 0,
|
|
},
|
|
"drift_items": drift_items,
|
|
}
|
|
|
|
|
|
def _recovery_package(drift_count=0):
|
|
actions = []
|
|
if drift_count:
|
|
actions = [
|
|
{
|
|
"action_id": "rollback-action-1",
|
|
"selector_id": "receipt-1",
|
|
"momo_icode": "MOMO-1",
|
|
"expected_pchome_id": "PCH-EXPECTED",
|
|
"actual_pchome_id": "PCH-DRIFT",
|
|
"status": "ready_for_controlled_reapply",
|
|
"rollback_sql_shape": "UPDATE pchome_product_matches SET pchome_id = :actual_pchome_id",
|
|
"controlled_reapply_sql_shape": "UPDATE pchome_product_matches SET pchome_id = :expected_pchome_id",
|
|
"selector_bindings": {
|
|
"momo_icode": "MOMO-1",
|
|
"expected_pchome_id": "PCH-EXPECTED",
|
|
"actual_pchome_id": "PCH-DRIFT",
|
|
},
|
|
"acceptance_gates": ["post_reapply_drift_verifier_returns_zero_drift"],
|
|
"writes_database": False,
|
|
}
|
|
]
|
|
return {
|
|
"result": (
|
|
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_PACKAGE_READY"
|
|
if drift_count
|
|
else "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_NOT_REQUIRED"
|
|
),
|
|
"summary": {
|
|
"drift_count": drift_count,
|
|
"drift_recovery_action_count": len(actions),
|
|
"drift_reapply_ready_count": len(actions),
|
|
"recovery_artifact_hash_match_count": 1,
|
|
"writes_database_count": 0,
|
|
},
|
|
"recovery_actions": actions,
|
|
"rollback_plan": {
|
|
"rollback_action_count": len(actions),
|
|
"executes_in_package": False,
|
|
"writes_database": False,
|
|
},
|
|
}
|
|
|
|
|
|
def _compact_package(drift_count=0):
|
|
return {
|
|
"result": (
|
|
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_DRIFT_REQUIRES_RECOVERY"
|
|
if drift_count
|
|
else "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_VERIFIED"
|
|
),
|
|
"summary": {
|
|
"target_selector_count": 2,
|
|
"post_apply_readback_pass_count": 2 - drift_count,
|
|
"drift_count": drift_count,
|
|
"drift_recovery_action_count": drift_count,
|
|
"compact_readback_artifact_hash_match_count": 1,
|
|
"writes_database_count": 0,
|
|
},
|
|
"compact_readback": {
|
|
"status": "blocked" if drift_count else "completed",
|
|
"next_machine_action": (
|
|
"run_controlled_reapply_check_mode"
|
|
if drift_count
|
|
else "keep_monitoring_drift"
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def _retention_package():
|
|
return {
|
|
"result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY_READY",
|
|
"success": True,
|
|
"summary": {
|
|
"artifact_count": 8,
|
|
"protected_active_chain_count": 4,
|
|
"retention_artifact_hash_match_count": 1,
|
|
"writes_database_count": 0,
|
|
},
|
|
}
|
|
|
|
|
|
def test_rollback_evidence_no_drift_builds_noop_machine_evidence():
|
|
package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package(
|
|
source_receipt_replay=_replay_package(),
|
|
source_drift_verifier=_drift_package(),
|
|
source_drift_recovery=_recovery_package(),
|
|
source_compact_readback=_compact_package(),
|
|
source_artifact_retention=_retention_package(),
|
|
)
|
|
|
|
assert package["policy"] == (
|
|
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence"
|
|
)
|
|
assert package["result"] == (
|
|
"DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED"
|
|
)
|
|
assert package["summary"]["controlled_apply_family_count"] == 5
|
|
assert package["summary"]["drift_count"] == 0
|
|
assert package["summary"]["rollback_required_count"] == 0
|
|
assert package["summary"]["rollback_evidence_ready_count"] == 1
|
|
assert package["summary"]["primary_human_gate_count"] == 0
|
|
assert package["rollback_evidence"]["status"] == "rollback_not_required"
|
|
assert package["rollback_actions"] == []
|
|
assert package["safety"]["executes_sql"] is False
|
|
assert package["safety"]["writes_database"] is False
|
|
assert package["all_checks_passed"] is True
|
|
|
|
|
|
def test_rollback_evidence_drift_ready_outputs_actions_without_db_write():
|
|
package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package(
|
|
source_receipt_replay=_replay_package(),
|
|
source_drift_verifier=_drift_package(drift_count=1),
|
|
source_drift_recovery=_recovery_package(drift_count=1),
|
|
source_compact_readback=_compact_package(drift_count=1),
|
|
source_artifact_retention=_retention_package(),
|
|
)
|
|
|
|
assert package["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_READY"
|
|
assert package["summary"]["drift_count"] == 1
|
|
assert package["summary"]["rollback_required_count"] == 1
|
|
assert package["summary"]["rollback_action_count"] == 1
|
|
assert package["summary"]["rollback_ready_action_count"] == 1
|
|
assert package["rollback_evidence"]["status"] == "rollback_ready"
|
|
assert package["rollback_evidence"]["next_machine_action"] == "run_controlled_reapply_check_mode"
|
|
assert package["rollback_actions"][0]["momo_icode"] == "MOMO-1"
|
|
assert "UPDATE pchome_product_matches" in package["rollback_actions"][0]["rollback_sql_shape"]
|
|
assert package["rollback_actions"][0]["executes_in_package"] is False
|
|
assert package["rollback_actions"][0]["writes_database"] is False
|
|
assert package["safety"]["executes_rollback"] is False
|
|
assert package["safety"]["executes_reapply"] is False
|
|
assert package["safety"]["writes_database"] is False
|
|
assert package["all_checks_passed"] is True
|
|
|
|
|
|
def test_rollback_evidence_route_uses_engine_and_returns_source_endpoint(monkeypatch):
|
|
from routes import ai_routes as routes
|
|
|
|
class FakeEngine:
|
|
disposed = False
|
|
|
|
def dispose(self):
|
|
self.disposed = True
|
|
|
|
fake_engine = FakeEngine()
|
|
monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", lambda _path: fake_engine)
|
|
|
|
def fake_builder(**kwargs):
|
|
assert kwargs["engine"] is fake_engine
|
|
return {
|
|
"success": True,
|
|
"policy": "read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence",
|
|
"result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED",
|
|
"summary": {"writes_database_count": 0, "primary_human_gate_count": 0},
|
|
"safety": {"writes_database": False},
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
"services.pchome_mapping_backlog_service.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package",
|
|
fake_builder,
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context(
|
|
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-rollback-evidence-package"
|
|
):
|
|
response = (
|
|
routes
|
|
.api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package
|
|
.__wrapped__()
|
|
)
|
|
|
|
payload = response.get_json()
|
|
assert fake_engine.disposed is True
|
|
assert payload["source_endpoint"] == (
|
|
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-artifact-retention-package"
|
|
)
|
|
assert payload["summary"]["primary_human_gate_count"] == 0
|
|
assert payload["safety"]["writes_database"] is False
|