Get Action as Text

/**
 * Reads the content of an action as text.
 *
 * @param {string} in_moduleName - Module which contains the action
 * @param {string} in_actionName - Action which contains the text
 * @returns {string} actionAsText
 *
 * @author Stefan Schnell <mail@stefan-schnell.de>
 * @license MIT
 * @version 0.4.1
 *
 * @example
 * var in_moduleName = "com.vmware.basic";
 * var in_actionName = "createDirectory";
 * System.log(
 *   System.getModule("de.stschnell").getActionAsText(
 *     in_moduleName, in_actionName
 *   )
 * );
 *
 * Checked with Aria Automation 8.12.0, 8.13.1, 8.14.1, 8.17.0, 8.18.1
 * and VCF Automation 9.0.0
 */

var _getActionAsText = {

  main : function(moduleName, actionName) {

    var allActions =
      System.getModule("com.vmware.library.action").getAllActions();

    var allActionsInModule = allActions.filter( function(actionItems) {
      return actionItems.module.name === moduleName;
    });
    if (allActionsInModule.length === 0) {
      throw new Error("No actions were found in the module " +
        moduleName + ".");
    }

    var action = allActionsInModule.filter( function(actionItem) {
      return actionItem.name === actionName;
    });
    if (action.length === 0) {
      throw new Error("The action " + actionName +
        " was not found in the module " + moduleName + ".");
    } else if (action.length > 1) {
      System.warn("Too many actions, with the name " + actionName +
        ", were found in the module " + moduleName + ".");
      action.forEach( function(actionItem) {
        System.warn("ID: " + actionItem.id);
      });
      throw new Error("Too many actions, with the name " + actionName +
        ", were found in the module " + moduleName + ".");
    }

    return action[0].script;

  }

};

if (
  in_moduleName && String(in_moduleName).trim() !== "" &&
  in_actionName && String(in_actionName).trim() !== ""
) {
  return _getActionAsText.main(
    String(in_moduleName).trim(),
    String(in_actionName).trim()
  );
} else {
  throw new Error(
    "in_moduleName or in_actionName argument can not be null"
  );
}

"""
@module de.stschnell

@version 0.1.0

@description Reads the content of an action as text.

@runtime python:3.10

@outputType Properties
"""

"""
@function getActionAsText

@param {string} in_moduleName - Module name which contains the action
@param {string} in_actionName - Action name which contains the text
@returns {Properties}

@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 0.1.0

Checked with Aria Automation 8.18.1
"""

import http
import json
import ssl
import urllib


def getActionId(
    vcoUrl: str,
    bearerToken: str,
    actionModule: str,
    actionName: str
) -> str:

    returnValue: str = ""
    response: http.client.HTTPResponse | None = None

    try:

        if (
            not vcoUrl
            or not vcoUrl.strip()
            or not bearerToken
            or not bearerToken.strip()
        ):
            raise ValueError("vcoUrl or bearerToken argument "
                             "can not be undefined or null")

        if (
            not actionModule
            or not actionModule.strip()
            or not actionName
            or not actionName.strip()
        ):
            raise ValueError("actionModule or actionName argument "
                             "can not be undefined or null")

        request: urllib.request.Request = urllib.request.Request(
            url = vcoUrl + "/api/actions",
            method = "GET"
        )

        request.add_header(
            "Authorization", "Bearer " + bearerToken
        )
        request.add_header(
            "Content-Type", "application/json;charset=utf-8"
        )
        request.add_header(
            "Accept", "application/json"
        )

        response = urllib.request.urlopen(
            request,
            context = ssl._create_unverified_context()
        )

        if response.getcode() != 200:
            raise RuntimeError(f"HTTP error: {response.getcode()}")

        actions: dict = json.loads(
            response.read().decode("utf-8")
        )

        found: bool = False
        action: dict
        for action in actions["link"]:
            attribute: dict
            for attribute in action["attributes"]:
                if attribute["name"] == "fqn" and \
                attribute["value"] == actionModule + "/" + actionName:
                    for attribute in action["attributes"]:
                        if attribute["name"] == "id":
                            returnValue = attribute["value"]
                            found = True
            if found:
                break

    except Exception as err:
        raise ValueError(
            f"An error occurred at get Action ID - {err}"
        ) from err

    finally:
        if response:
            response.close()

    return returnValue


def getActionAsText(
    vcoUrl: str,
    bearerToken: str,
    actionModule: str,
    actionName: str
) -> str:

    returnValue: str = ""
    response: http.client.HTTPResponse | None = None

    try:

        actionId: str = getActionId(
            vcoUrl,
            bearerToken,
            actionModule,
            actionName
        )

        if actionId:

            request: urllib.request.Request = urllib.request.Request(
                url = vcoUrl + "/api/actions/" + actionId,
                method = "GET"
            )

            request.add_header(
                "Authorization", "Bearer " + bearerToken
            )
            request.add_header(
                "Content-Type", "application/json;charset=utf-8"
            )
            request.add_header(
                "Accept", "application/json"
            )

            response = urllib.request.urlopen(
                request,
                context = ssl._create_unverified_context()
            )

            if response.getcode() != 200:
                raise RuntimeError(f"HTTP error: {response.getcode()}")

            actionDetails: dict = json.loads(
                response.read().decode("utf-8")
            )
            if actionDetails.get("script"):
                returnValue = actionDetails["script"]

    except Exception as err:
        raise ValueError(
            f"An error occurred at retrieve action definition - {err}"
        ) from err

    finally:
        if response:
            response.close()

    return returnValue


def handler(context: dict, inputs: dict) -> dict:

    vcoUrl: str = context["vcoUrl"]
    bearerToken: str = context["getToken"]()

    # actionModule: str = inputs["in_moduleName"]
    # actionName: str = inputs["in_actionName"]

    actionModule: str = "com.vmware.basic"
    actionName: str = "createDirectory"

    result: str = ""
    outputs: dict = {}

    try:

        result = getActionAsText(
            vcoUrl,
            bearerToken,
            actionModule,
            actionName
        )

        outputs = {
            "status": "done",
            "error": None,
            "result": result
        }

    except Exception as err:

        outputs = {
            "status": "incomplete",
            "error": repr(err),
            "result": None
        }

    return outputs

def getActionAsText(
    vcoUrl: str,
    bearerToken: str,
    moduleName: str,
    actionName: str
) -> str:
    """ Delivers the content of an action as text.

    @param vcoUrl {string} - URL of Aria orchestrator
    @param bearerToken {string}
    @param moduleName {string} - Module of the action
    @param actionName {string} - Name of the action
    @returns {string}

    @author Stefan Schnell <mail@stefan-schnell.de>
    @license MIT
    @version 0.1.0
    """

    if (
        not moduleName
        or not moduleName.strip()
        or not actionName
        or not actionName.strip()
    ):
        raise ValueError("moduleName or actionName argument "
                         "can not be undefined or null")

    http = Http()
    result: str = ""

    try:

        actions: dict = http.request(
            url = vcoUrl + "/api/actions",
            bearerToken = bearerToken
        )

        if not actions:
            raise ValueError("No actions were detected")

        actionList: list = []

        for action in actions["link"]:

            fqn: str = ""

            for attribute in action["attributes"]:
                if attribute["name"] == "fqn":
                    fqn = attribute["value"]

            if fqn != "":
                actionList.append(
                    {"href": action["href"], "fqn": fqn}
                )

        actionList.sort(key = lambda x: x["fqn"])

        actionList = [
            x for x in actionList \
            if (moduleName.lower() + "/" + actionName.lower()) \
            in x["fqn"].lower()
        ]

        for action in actionList:

            actionDetails = http.request(
                url = action["href"],
                bearerToken = bearerToken
            )
            if not actionDetails:
                raise Exception("No action details were detected")

            if "script" in actionDetails:
                result = actionDetails["script"]

    except Exception as err:
        raise Exception("Error at getActionAsText - " + repr(err))

    return result