Validate all subscriptions using their corresponding validation workflow.
Validation workflows only run on subscriptions that are active, even when they could be run on provisioning
subscriptions. E.g. for routers, they can manually be validated when provisioning, but are not included in this
schedule.
Validation workflows will, in principle, only run on subscriptions that are in sync. Except when a specific
validation workflow is marked as usable while out of sync in the list WF_USABLE_WHILE_OUT_OF_SYNC
.
Source code in gso/schedules/validate_subscriptions.py
| @shared_task
@scheduler(CronScheduleConfig(name="Subscriptions Validator", minute="10", hour="3"))
def validate_subscriptions() -> None:
"""Validate all subscriptions using their corresponding validation workflow.
Validation workflows only run on subscriptions that are active, even when they could be run on provisioning
subscriptions. E.g. for routers, they can manually be validated when provisioning, but are not included in this
schedule.
Validation workflows will, in principle, only run on subscriptions that are in sync. Except when a specific
validation workflow is marked as usable while out of sync in the list `WF_USABLE_WHILE_OUT_OF_SYNC`.
"""
subscriptions = get_active_subscriptions()
if not subscriptions:
logger.info("No subscriptions to validate")
return
for subscription in subscriptions:
if re.search(r"SKIP VALIDATION: .+", subscription.note or ""):
# The subscription is marked to skip validation altogether. We continue to the next subscription.
# FIXME: This is a temporary hack and MUST be removed!
logger.warning(
"Manually skipped validation workflows for a subscription.",
product=subscription.product.name,
subscription_id=subscription.subscription_id,
subscription_description=subscription.description,
skip_reason=subscription.note,
)
continue
validation_workflows = get_validation_product_workflows_for_subscription(subscription)
for workflow in validation_workflows:
validation_workflow_usable = (subscription.status in TARGET_DEFAULT_USABLE_MAP[Target.VALIDATE]) and (
subscription.insync or (workflow in WF_USABLE_WHILE_OUT_OF_SYNC)
)
if validation_workflow_usable:
logger.info(
"Found a usable validation workflow, scheduling task.",
product=subscription.product.name,
subscription_id=subscription.subscription_id,
subscription_description=subscription.description,
workflow=workflow,
)
json = [{"subscription_id": str(subscription.subscription_id)}]
validate_func = get_execution_context()["validate"]
validate_func(workflow, json=json)
else:
logger.info(
"Validation workflow is not usable on this subscription instance",
product=subscription.product.name,
subscription_id=subscription.subscription_id,
subscription_description=subscription.description,
status=subscription.status,
insync=subscription.insync,
workflow=workflow,
)
if not validation_workflows:
logger.warning(
"SubscriptionTable has no validation workflow",
product=subscription.product.name,
subscription_id=subscription.subscription_id,
subscription_description=subscription.description,
)
|