refactor: simplify Campaign editor validation

This commit is contained in:
2026-07-21 18:16:09 +02:00
parent 0ac903c82d
commit 03100b77db

View File

@@ -29,6 +29,47 @@ class CampaignMailProfileBoundaryError(ValueError):
"""
def _validated_opt_ins(value: Any) -> dict[str, bool]:
if not isinstance(value, dict) or any(
key not in CAMPAIGN_OPT_IN_KEYS for key in value
):
raise CampaignMailProfileBoundaryError(
"Campaign editor opt_ins contains unsupported fields"
)
if any(not isinstance(item, bool) for item in value.values()):
raise CampaignMailProfileBoundaryError(
"Campaign editor opt_ins values must be booleans"
)
return copy.deepcopy(value)
def _is_valid_field_override(key: Any, value: Any) -> bool:
return (
isinstance(key, str)
and bool(key.strip())
and len(key) <= 256
and isinstance(value, bool)
)
def _validated_field_overrides(value: Any) -> dict[str, bool]:
if not isinstance(value, dict) or len(value) > 10_000:
raise CampaignMailProfileBoundaryError(
"Campaign editor field_overrides must be a bounded object"
)
if any(not _is_valid_field_override(key, item) for key, item in value.items()):
raise CampaignMailProfileBoundaryError(
"Campaign editor field_overrides must map short field names to booleans"
)
return copy.deepcopy(value)
def _validated_required_string(value: Any, *, max_length: int, error: str) -> str:
if not isinstance(value, str) or not value.strip() or len(value) > max_length:
raise CampaignMailProfileBoundaryError(error)
return value.strip()
def validate_campaign_editor_state(
value: dict[str, Any] | None,
*,
@@ -55,32 +96,17 @@ def validate_campaign_editor_state(
result: dict[str, Any] = {}
if "created_from" in value:
created_from = value["created_from"]
if not isinstance(created_from, str) or not created_from.strip() or len(created_from) > 128:
raise CampaignMailProfileBoundaryError("Campaign editor created_from must be a short string")
result["created_from"] = created_from.strip()
result["created_from"] = _validated_required_string(
value["created_from"],
max_length=128,
error="Campaign editor created_from must be a short string",
)
if "opt_ins" in value:
opt_ins = value["opt_ins"]
if not isinstance(opt_ins, dict) or any(key not in CAMPAIGN_OPT_IN_KEYS for key in opt_ins):
raise CampaignMailProfileBoundaryError("Campaign editor opt_ins contains unsupported fields")
if any(not isinstance(item, bool) for item in opt_ins.values()):
raise CampaignMailProfileBoundaryError("Campaign editor opt_ins values must be booleans")
result["opt_ins"] = copy.deepcopy(opt_ins)
result["opt_ins"] = _validated_opt_ins(value["opt_ins"])
if "field_overrides" in value:
overrides = value["field_overrides"]
if not isinstance(overrides, dict) or len(overrides) > 10_000:
raise CampaignMailProfileBoundaryError("Campaign editor field_overrides must be a bounded object")
if any(
not isinstance(key, str)
or not key.strip()
or len(key) > 256
or not isinstance(item, bool)
for key, item in overrides.items()
):
raise CampaignMailProfileBoundaryError(
"Campaign editor field_overrides must map short field names to booleans"
)
result["field_overrides"] = copy.deepcopy(overrides)
result["field_overrides"] = _validated_field_overrides(
value["field_overrides"]
)
if "review_send" in value:
result["review_send"] = _validated_server_review_state(value["review_send"])
return result
@@ -122,34 +148,62 @@ def campaign_editor_state_for_edit(value: Any) -> dict[str, Any]:
return state
def _is_valid_reviewed_message_key(value: Any) -> bool:
return isinstance(value, str) and bool(value.strip()) and len(value) <= 512
def _validated_reviewed_message_keys(value: Any) -> list[str]:
if not isinstance(value, list) or len(value) > 100_000:
raise CampaignMailProfileBoundaryError(
"Campaign reviewed message keys are invalid"
)
if any(not _is_valid_reviewed_message_key(key) for key in value):
raise CampaignMailProfileBoundaryError(
"Campaign reviewed message keys are invalid"
)
return list(dict.fromkeys(key.strip() for key in value))
def _validated_review_actor(value: Any) -> str | None:
if value is not None and (not isinstance(value, str) or len(value) > 256):
raise CampaignMailProfileBoundaryError("Campaign review actor is invalid")
return value
def _validated_server_review_state(value: Any) -> dict[str, Any]:
if not isinstance(value, dict) or any(key not in CAMPAIGN_REVIEW_STATE_KEYS for key in value):
raise CampaignMailProfileBoundaryError("Campaign review editor state is invalid")
if not isinstance(value, dict) or any(
key not in CAMPAIGN_REVIEW_STATE_KEYS for key in value
):
raise CampaignMailProfileBoundaryError(
"Campaign review editor state is invalid"
)
build_token = value.get("build_token")
inspected = value.get("inspection_complete")
keys = value.get("reviewed_message_keys", [])
updated_at = value.get("updated_at")
updated_by = value.get("updated_by_user_id")
if not isinstance(build_token, str) or not build_token.strip() or len(build_token) > 256:
raise CampaignMailProfileBoundaryError("Campaign review build token is invalid")
validated_build_token = _validated_required_string(
build_token,
max_length=256,
error="Campaign review build token is invalid",
)
if not isinstance(inspected, bool):
raise CampaignMailProfileBoundaryError("Campaign review completion state is invalid")
if (
not isinstance(keys, list)
or len(keys) > 100_000
or any(not isinstance(key, str) or not key.strip() or len(key) > 512 for key in keys)
):
raise CampaignMailProfileBoundaryError("Campaign reviewed message keys are invalid")
if not isinstance(updated_at, str) or not updated_at.strip() or len(updated_at) > 128:
raise CampaignMailProfileBoundaryError("Campaign review timestamp is invalid")
if updated_by is not None and (not isinstance(updated_by, str) or len(updated_by) > 256):
raise CampaignMailProfileBoundaryError("Campaign review actor is invalid")
raise CampaignMailProfileBoundaryError(
"Campaign review completion state is invalid"
)
validated_keys = _validated_reviewed_message_keys(keys)
validated_updated_at = _validated_required_string(
updated_at,
max_length=128,
error="Campaign review timestamp is invalid",
)
validated_updated_by = _validated_review_actor(updated_by)
return {
"build_token": build_token.strip(),
"build_token": validated_build_token,
"inspection_complete": inspected,
"reviewed_message_keys": list(dict.fromkeys(key.strip() for key in keys)),
"updated_at": updated_at.strip(),
"updated_by_user_id": updated_by,
"reviewed_message_keys": validated_keys,
"updated_at": validated_updated_at,
"updated_by_user_id": validated_updated_by,
}