INNER CODE UNIT · Python
validate_plan
yaojingang/GEOFlow · deployment/generate-upgrade-plan.py:28
def validate_plan(plan: dict) -> None:
exact_fields(plan, {"schema_version", "strategy", "allowed_sources", "migrations", "compatibility", "steps"}, "Upgrade plan")
if type(plan["schema_version"]) is not int or plan["schema_version"] != 1 or plan["strategy"] not in ("maintenance", "online"):
raise ValueError("Upgrade plan schema or strategy is unsupported")
sources = plan["allowed_sources"]
if type(sources) is not list or any(type(source) is not int or source < 1 or source > 2**63 - 1 for source in sources) or len(sources) != len(set(sources)):
raise ValueError("Upgrade source sequences must be unique positive application integers")
compatible = plan["compatibility"]
exact_fields(compatible, {"schema", "queue", "cache", "storage"}, "Compatibility")
if any(type(value) is not bool for value in compatible.values()):
raise ValueError("Compatibility fields must be booleans")
if type(plan["migrations"]) is not list or type(plan["steps"]) is not list:
raise ValueError("Migrations and steps must be arrays")
names = set()
for migration in plan["migrations"]:
exact_fields(migration, {"name", "sha256", "online"}, "Migration")
if type(migration["name"]) is not str or not re.fullmatch(r"[0-9]{4}_[0-9]{2}_[0-9]{2}_[0-9]{6}_[A-Za-z0-9_]+", migration["name"]) or migration["name"] in names:
raise ValueError("Migration identity is invalid or duplicated")