def upgrade()

in src/sfctl/custom_app.py [0:0]

• 46 lines of code
• 2 McCabe index (conditional complexity)

def upgrade(  # pylint: disable=too-many-arguments,too-many-locals,missing-docstring
        client, application_id, application_version, parameters,
        mode="UnmonitoredAuto", replica_set_check_timeout=None,
        force_restart=None, failure_action=None,
        health_check_wait_duration="0",
        health_check_stable_duration="PT0H2M0S",
        health_check_retry_timeout="PT0H10M0S",
        upgrade_timeout="P10675199DT02H48M05.4775807S",
        upgrade_domain_timeout="P10675199DT02H48M05.4775807S",
        warning_as_error=False,
        max_unhealthy_apps=0, default_service_health_policy=None,
        service_health_policy=None, timeout=60):
    from azure.servicefabric.models import (ApplicationUpgradeDescription,
                                            MonitoringPolicyDescription,
                                            ApplicationHealthPolicy)

    from sfctl.custom_health import (parse_service_health_policy_map,
                                     parse_service_health_policy)

    monitoring_policy = MonitoringPolicyDescription(
        failure_action=failure_action,
        health_check_wait_duration_in_milliseconds=health_check_wait_duration,
        health_check_stable_duration_in_milliseconds=health_check_stable_duration,
        health_check_retry_timeout_in_milliseconds=health_check_retry_timeout,
        upgrade_timeout_in_milliseconds=upgrade_timeout,
        upgrade_domain_timeout_in_milliseconds=upgrade_domain_timeout
    )

    # Must always have empty list
    app_params = parse_app_params(parameters)
    if app_params is None:
        app_params = []

    def_shp = parse_service_health_policy(default_service_health_policy)

    map_shp = parse_service_health_policy_map(service_health_policy)

    app_health_policy = ApplicationHealthPolicy(
        consider_warning_as_error=warning_as_error,
        max_percent_unhealthy_deployed_applications=max_unhealthy_apps,
        default_service_type_health_policy=def_shp,
        service_type_health_policy_map=map_shp)

    desc = ApplicationUpgradeDescription(
        name='fabric:/' + application_id,
        target_application_type_version=application_version,
        parameters=app_params,
        upgrade_kind='Rolling',
        rolling_upgrade_mode=mode,
        upgrade_replica_set_check_timeout_in_seconds=replica_set_check_timeout,
        force_restart=force_restart,
        monitoring_policy=monitoring_policy,
        application_health_policy=app_health_policy)

    client.start_application_upgrade(application_id, desc, timeout)