provisioning_proxy.py

"""The Provisioning Proxy service, which interacts with LSO running externally.

LSO is responsible for executing Ansible playbooks, that deploy subscriptions.
"""
import json
import logging

import requests
from orchestrator import inputstep
from orchestrator.config.assignee import Assignee
from orchestrator.domain import SubscriptionModel
from orchestrator.forms import FormPage, ReadOnlyField
from orchestrator.forms.validators import Accept, Label, LongText
from orchestrator.types import FormGenerator, State, UUIDstr, strEnum
from orchestrator.utils.json import json_dumps
from pydantic import validator

from gso import settings
from gso.products.product_types.device import DeviceProvisioning
from gso.products.product_types.iptrunk import Iptrunk, IptrunkProvisioning

logger = logging.getLogger(__name__)


class CUDOperation(strEnum):
    """Enum for different C(R)UD operations that the provisioning proxy supports.

    Read is not applicable, hence these become CUD and not CRUD operations.
    """

    #: Creation is done with a POST request
    POST = "POST"
    #: Updating is done with a PUT request
    PUT = "PUT"
    #: Removal is done with a DELETE request
    DELETE = "DELETE"


def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr, operation: CUDOperation) -> None:
    """Send a request to LSO. The callback address is derived using the process ID provided.

    :param str endpoint: The LSO-specific endpoint to call, depending on the type of service object that is acted upon.
    :param dict parameters: JSON body for the request, which will almost always at least consist of a subscription
        object, and a boolean value to indicate a dry run.
    :param UUIDstr process_id: The process ID that this request is a part of, used to call back to when the execution
        of the playbook is completed.
    :param :class:`CUDOperation` operation: The specific operation that is performed with the request.
    """
    oss = settings.load_oss_params()
    pp_params = oss.PROVISIONING_PROXY
    assert pp_params

    callback_url = f"{settings.load_oss_params().GENERAL.public_hostname}" f"/api/processes/{process_id}/resume"
    logger.debug("[provisioning proxy] provisioning for process %s", process_id)

    parameters.update({"callback": callback_url})
    url = f"{pp_params.scheme}://{pp_params.api_base}/api/{endpoint}"

    request = None

    if operation == CUDOperation.POST:
        request = requests.post(url, json=parameters, timeout=10000)
    elif operation == CUDOperation.PUT:
        request = requests.put(url, json=parameters, timeout=10000)
    elif operation == CUDOperation.DELETE:
        request = requests.delete(url, json=parameters, timeout=10000)

    if request.status_code != 200:
        logger.debug(request.content)
        raise AssertionError(request.content)


def provision_device(subscription: DeviceProvisioning, process_id: UUIDstr, dry_run: bool = True) -> None:
    """Provision a new device using LSO.

    :param :class:`DeviceProvisioning` subscription: The subscription object
        that is to be provisioned.
    :param UUIDstr process_id: The related process ID, used for callback.
    :param bool dry_run: A boolean indicating whether this should be a dry run
        or not, defaults to ``True``.
    """
    parameters = {"dry_run": dry_run, "subscription": json.loads(json_dumps(subscription))}

    _send_request("device", parameters, process_id, CUDOperation.POST)


def provision_ip_trunk(
    subscription: IptrunkProvisioning, process_id: UUIDstr, config_object: str, dry_run: bool = True
) -> None:
    """Provision an IP trunk service using LSO.

    :param :class:`IptrunkProvisioning` subscription: The subscription object
        that is to be provisioned.
    :param UUIDstr process_id: The related process ID, used for callback.
    :param str config_object: The type of object that is deployed
    :param bool dry_run: A boolean indicating whether this should be a dry run
        or not, defaults to ``True``.
    """
    parameters = {
        "subscription": json.loads(json_dumps(subscription)),
        "dry_run": dry_run,
        "verb": "deploy",
        "object": config_object,
    }

    _send_request("ip_trunk", parameters, process_id, CUDOperation.POST)


# def modify_ip_trunk(old_subscription: Iptrunk,
#                    new_subscription: Iptrunk,
#                    process_id: UUIDstr,
#                    dry_run: bool = True):
#    """
#    Function that modifies an existing IP trunk subscription using LSO.
#
#    :param :class:`Iptrunk` old_subscription: The subscription object, before
#        its modification.
#    :param :class:`Iptrunk` new_subscription: The subscription object, after
#        modifications have been made to it.
#    :param UUIDstr process_id: The related process ID, used for callback.
#    :param bool dry_run: A boolean indicating whether this should be a dry ryn
#        or not, defaults to ``True``.
#    """
#    parameters = {
#        'dry_run': dry_run,
#        'old_subscription': old_subscription,
#        'subscription': new_subscription
#        # ... missing parameters
#    }
#
#    _send_request('ip_trunk', parameters, process_id, CUDOperation.PUT)


def deprovision_ip_trunk(subscription: Iptrunk, process_id: UUIDstr, dry_run: bool = True) -> None:
    """Deprovision an IP trunk service using LSO.

    :param :class:`IptrunkProvisioning` subscription: The subscription object
        that is to be provisioned.
    :param UUIDstr process_id: The related process ID, used for callback.
    :param bool dry_run: A boolean indicating whether this should be a dry run
        or not, defaults to ``True``.
    """
    parameters = {"subscription": json.loads(json_dumps(subscription)), "dry_run": dry_run, "verb": "terminate"}

    _send_request("ip_trunk", parameters, process_id, CUDOperation.DELETE)


@inputstep("Await provisioning proxy results", assignee=Assignee("SYSTEM"))
def await_pp_results(subscription: SubscriptionModel, label_text: str) -> FormGenerator:
    class ProvisioningResultPage(FormPage):
        class Config:
            title = f"Deploying {subscription.product.name}..."

        warning_label: Label = label_text  # type: ignore
        pp_run_results: dict = None  # type: ignore
        confirm: Accept = Accept("INCOMPLETE")

        @validator("pp_run_results", allow_reuse=True, pre=True, always=True)
        def run_results_must_be_given(cls, run_results: dict) -> dict | None:
            if run_results is None:
                raise ValueError("Run results may not be empty. Wait for the provisioning proxy to finish.")
            return run_results

    result_page = yield ProvisioningResultPage

    return result_page.dict()


@inputstep("Confirm provisioning proxy results", assignee=Assignee("SYSTEM"))
def confirm_pp_results(state: State) -> FormGenerator:
    class ConfirmRunPage(FormPage):
        class Config:
            title = (
                f"Execution for "
                f"{state['subscription']['product']['name']} "
                f"completed, please confirm the results below."
            )

        run_status: str = ReadOnlyField(state["pp_run_results"]["status"])
        run_results: LongText = ReadOnlyField(f"{state['pp_run_results']['output']}")
        confirm: Accept = Accept("INCOMPLETE")

    yield ConfirmRunPage

    return state