def governance_service_deploy_cmd()

in src/tools/azure-cli-extension/cleanroom/azext_cleanroom/custom.py [0:0]

• 115 lines of code
• 24 McCabe index (conditional complexity)

def governance_service_deploy_cmd(cmd, gov_client_name):
    cgs_endpoint = get_cgs_client_endpoint(cmd, gov_client_name)

    # Download the constitution and js_app to deploy.
    dir_path = os.path.dirname(os.path.realpath(__file__))
    bin_folder = os.path.join(dir_path, "bin")
    if not os.path.exists(bin_folder):
        os.makedirs(bin_folder)

    constitution, bundle = download_constitution_jsapp(bin_folder)

    # Submit and accept set_constitution proposal.
    logger.warning("Deploying constitution on CCF")
    content = {
        "actions": [
            {
                "name": "set_constitution",
                "args": {"constitution": constitution},
            }
        ]
    }
    r = requests.post(f"{cgs_endpoint}/proposals/create", json=content)
    if r.status_code != 200:
        raise CLIError(
            f"set_constitution proposal failed with status: {r.status_code} and response: {r.text}"
        )

    # A set_constitution proposal might already be accepted if the default constitution was
    # unconditionally accepting proposals. So only vote if not already accepted.
    if r.json()["proposalState"] != "Accepted":
        proposal_id = r.json()["proposalId"]
        r = requests.post(f"{cgs_endpoint}/proposals/{proposal_id}/ballots/vote_accept")
        if r.status_code != 200:
            raise CLIError(
                f"set_constitution proposal acceptance failed with status: {r.status_code} and "
                + f"response: {r.text}"
            )
        if r.json()["proposalState"] == "Open":
            logger.warning(
                "set_constitution proposal %s remains open. "
                + "Other members need to vote their acceptance for changes to take affect.",
                proposal_id,
            )
        elif r.json()["proposalState"] == "Rejected":
            raise CLIError(f"set_constitution proposal {proposal_id} was rejected")

    # Submit and accept set_js_runtime_options proposal.
    logger.warning("Configuring js runtime options on CCF")
    content = {
        "actions": [
            {
                "name": "set_js_runtime_options",
                "args": {
                    "max_heap_bytes": 104857600,
                    "max_stack_bytes": 1048576,
                    "max_execution_time_ms": 1000,
                    "log_exception_details": True,
                    "return_exception_details": True,
                },
            }
        ]
    }
    r = requests.post(
        f"{cgs_endpoint}/proposals/create", json=content
    )  # [missing-timeout]
    if r.status_code != 200:
        raise CLIError(
            f"set_js_runtime_options proposal failed with status: {r.status_code} and response: {r.text}"
        )

    proposal_id = r.json()["proposalId"]
    r = requests.post(f"{cgs_endpoint}/proposals/{proposal_id}/ballots/vote_accept")
    if r.status_code != 200:
        raise CLIError(
            f"set_js_runtime_options proposal acceptance failed with status: {r.status_code} "
            + f"and response: {r.text}"
        )

    # Submit and accept set_js_app proposal.
    logger.warning("Deploying governance service js application on CCF")
    content = {
        "actions": [
            {
                "name": "set_js_app",
                "args": {"bundle": bundle},
            }
        ]
    }
    r = requests.post(f"{cgs_endpoint}/proposals/create", json=content)
    if r.status_code != 200:
        raise CLIError(
            f"set_js_app proposal failed with status: {r.status_code} and response: {r.text}"
        )

    proposal_id = r.json()["proposalId"]
    r = requests.post(f"{cgs_endpoint}/proposals/{proposal_id}/ballots/vote_accept")
    if r.status_code != 200:
        raise CLIError(
            f"set_js_app proposal acceptance failed with status: {r.status_code} and response: {r.text}"
        )
    if r.json()["proposalState"] == "Open":
        logger.warning(
            "set_js_app proposal %s remains open. "
            + "Other members need to vote their acceptance for changes to take affect.",
            proposal_id,
        )
    elif r.json()["proposalState"] == "Rejected":
        raise CLIError(f"set_js_app proposal {proposal_id} was rejected")

    # Enable the OIDC issuer by default as its required for mainline scenarios.
    r = governance_oidc_issuer_show_cmd(cmd, gov_client_name)
    if r["enabled"] != True:
        logger.warning("Enabling OIDC Issuer capability")
        r = governance_oidc_issuer_propose_enable_cmd(cmd, gov_client_name)
        proposal_id = r["proposalId"]
        r = requests.post(f"{cgs_endpoint}/proposals/{proposal_id}/ballots/vote_accept")
        if r.status_code != 200:
            raise CLIError(
                f"enable_oidc_issuer proposal acceptance failed with status: {r.status_code} and response: {r.text}"
            )
        if r.json()["proposalState"] == "Open":
            logger.warning(
                "enable_oidc_issuer proposal %s remains open. "
                + "Other members need to vote their acceptance for changes to take affect.",
                proposal_id,
            )
        elif r.json()["proposalState"] == "Rejected":
            raise CLIError(f"enable_oidc_issuer proposal {proposal_id} was rejected")

        governance_oidc_issuer_generate_signing_key_cmd(cmd, gov_client_name)
    else:
        logger.warning("OIDC Issuer capability is already enabled")