Skip to content

Validate switch

Validation workflow for switch subscription objects.

check_netbox_device(subscription)

Fetch the device in Netbox. Will raise an exception if it is not found.

Source code in gso/workflows/switch/validate_switch.py 
15
16
17
18
@step("Validate switch in Netbox")
def check_netbox_device(subscription: Switch) -> None:
    """Fetch the device in Netbox. Will raise an exception if it is not found."""
    NetboxClient().get_device_by_name(subscription.switch.fqdn)

verify_base_config(subscription)

Workflow step for running a playbook that checks whether base config has drifted.

Source code in gso/workflows/switch/validate_switch.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
@step("Check base config for drift")
def verify_base_config(subscription: dict[str, Any]) -> LSOState:
    """Workflow step for running a playbook that checks whether base config has drifted."""
    return {
        "playbook_name": "switch_base_config.yaml",
        "inventory": {"all": {"hosts": {subscription["switch"]["fqdn"]: None}}},
        "extra_vars": {
            "subscription_json": subscription,
            "verb": "deploy",
            "dry_run": "true",
            "is_verification_workflow": "true",
        },
    }

validate_switch()

Validate an existing, active switch subscription.

Source code in gso/workflows/switch/validate_switch.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@workflow(
    "Validate switch subscription", target=Target.SYSTEM, initial_input_form=(wrap_modify_initial_input_form(None))
)
def validate_switch() -> StepList:
    """Validate an existing, active switch subscription."""
    return (
        begin
        >> store_process_subscription(Target.SYSTEM)
        >> unsync
        >> check_netbox_device
        >> anonymous_lso_interaction(verify_base_config)
        >> resync
        >> done
    )