"""
Tests an action via Orchestrator REST API
@author Stefan Schnell <mail@stefan-schnell.de>
@license MIT
@version 0.1.0
@param {string} in_actionModule - Name of the module
@param {string} in_actionName - Name of the action
@param {string} in_outputType - Data type of output, default Properties
@param {string} in_script - Script code
@param {string} in_version - Version of the action, default 0.1.0
@param {string} in_runtime - Name of the runtime, if null JavaScript
@returns {Properties}
@example
System.getModule("de.stschnell").testAction(
"de.stschnell",
"helloWorld",
"string",
"System.log(\"Hello World\");\n" +
"return \"Hello World\"",
null,
null
);
@example
System.getModule("de.stschnell").testAction(
"de.stschnell",
"helloWorld",
null,
"import json\n" +
"def handler(context, inputs):\n" +
" print(\"Hello World\")\n" +
" return {\"result\": \"Hello World\"}",
null,
"python:3.10"
);
Checked with Aria Automation 8.18.1
"""
import ast
import json
import ssl
import time
import urllib.error
import urllib.request
def createAction(
vcoUrl: str,
bearerToken: str,
actionModule: str,
actionName: str,
outputType: str | None,
script: str,
version: str | None,
runtime: str | None
) -> dict:
""" Creates an action in the Orchestrator
"""
try:
if not (vcoUrl and vcoUrl.strip()) or \
not (bearerToken and bearerToken.strip()):
raise ValueError("vcoUrl or bearerToken argument "
"cannot be undefined or null")
if not (actionModule and actionModule.strip()) or \
not (actionName and actionName.strip()):
raise ValueError("actionModule or actionName argument "
"cannot be undefined or null")
_outputType: str = outputType \
if (outputType and outputType.strip()) else "Properties"
if not script or not script.strip():
raise ValueError("script argument "
"cannot be undefined or null")
_version: str = version \
if (version and version.strip()) else "0.1.0"
url = f"{vcoUrl.rstrip('/')}/api/actions"
body: dict = {
"input-parameters": [],
"module": actionModule,
"name": actionName,
"output-type": _outputType,
"script": script,
"version": _version
}
if runtime and runtime.strip():
body["runtime"] = runtime
request = urllib.request.Request(
url = url,
method = "POST",
data = json.dumps(body).encode("utf-8")
)
request.add_header("Content-Type", "application/json")
request.add_header("Accept", "application/json, text/plain, */*")
request.add_header("Authorization", f"Bearer {bearerToken}")
response = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
responseCode: int = response.status
if responseCode == 201:
try:
return json.loads(response.read().decode("utf-8"))
except json.JSONDecodeError as jsonErr:
raise RuntimeError(f"Invalid JSON response: {jsonErr}")
else:
raise RuntimeError(f"Unexpected HTTP status code: {responseCode}")
except urllib.error.HTTPError as err:
errorDetails = err.read().decode("utf-8")
raise RuntimeError(f"HTTP error {err.code} at createAction: "
f"{errorDetails}") from err
except Exception as err:
raise RuntimeError(f"Error at createAction: {err}") from err
def executeAction(
vcoUrl: str,
bearerToken: str,
actionId: str
) -> dict:
""" Invokes an action in the Orchestrator
"""
try:
if not (vcoUrl and vcoUrl.strip()) or \
not (bearerToken and bearerToken.strip()):
raise ValueError("vcoUrl or bearerToken argument "
"cannot be undefined or null")
if not actionId or not actionId.strip():
raise ValueError("actionId argument "
"cannot be undefined or null")
url = f"{vcoUrl.rstrip('/')}/api/actions/{actionId}/executions"
body: dict = {
"async-execution": False,
"parameters": []
}
data = bytes(json.dumps(body).encode("utf-8"))
request = urllib.request.Request(
url = url,
method = "POST",
data = data
)
request.add_header("Content-Type", "application/json")
request.add_header("Accept", "application/json, text/plain, */*")
request.add_header("Authorization", f"Bearer {bearerToken}")
response = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
responseCode: int = response.status
if responseCode == 200:
try:
return json.loads(response.read().decode("utf-8"))
except json.JSONDecodeError as jsonErr:
raise RuntimeError(f"Invalid JSON response: {jsonErr}")
else:
raise RuntimeError(f"Unexpected HTTP status code: {responseCode}")
except urllib.error.HTTPError as err:
errorDetails = err.read().decode("utf-8")
raise RuntimeError(f"HTTP Error {err.code} at executeAction: "
f"{errorDetails}") from err
except Exception as err:
raise RuntimeError(f"Error at executeAction: {err}") from err
def getActionLog(
vcoUrl: str,
bearerToken: str,
executionId: str
) -> dict:
""" Delivers the content of the action log
"""
try:
if not (vcoUrl and vcoUrl.strip()) or \
not (bearerToken and bearerToken.strip()):
raise ValueError("vcoUrl or bearerToken argument "
"cannot be undefined or null")
if not executionId or not executionId.strip():
raise ValueError("executionId argument "
"cannot be undefined or null")
url = (f"{vcoUrl.rstrip('/')}/api/actions/{executionId}"
"/logs?maxResult=2147483647")
request = urllib.request.Request(
url = url,
method = "GET"
)
request.add_header("Accept", "application/json, text/plain, */*")
request.add_header("Authorization", f"Bearer {bearerToken}")
response = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
responseCode: int = response.status
if responseCode == 200:
rawData = response.read().decode("utf-8")
try:
return json.loads(rawData)
except json.JSONDecodeError as jsonErr:
try:
return ast.literal_eval(rawData)
except (ValueError, SyntaxError) as astErr:
return {"logRawData": rawData}
else:
return {}
except urllib.error.HTTPError as err:
errorDetails = err.read().decode("utf-8")
raise RuntimeError(f"HTTP Error {err.code} at getActionLog: "
f"{errorDetails}") from err
except Exception as err:
print(f"Error at getActionLog: {err}")
return {}
def deleteAction(
vcoUrl: str,
bearerToken: str,
actionId: str
) -> bool:
""" Removes an action in the Orchestrator
"""
try:
if not (vcoUrl and vcoUrl.strip()) or \
not (bearerToken and bearerToken.strip()):
raise ValueError("vcoUrl or bearerToken argument "
"cannot be undefined or null")
if not actionId or not actionId.strip():
raise ValueError("actionId argument "
"cannot be undefined or null")
url = f"{vcoUrl.rstrip('/')}/api/actions/{actionId}?force=false"
request = urllib.request.Request(
url = url,
method = "DELETE"
)
request.add_header("Accept", "application/json, text/plain, */*")
request.add_header("Authorization", f"Bearer {bearerToken}")
response = urllib.request.urlopen(
request,
context = ssl._create_unverified_context()
)
responseCode: int = response.status
if responseCode == 200:
return True
else:
return False
except Exception as err:
print(f"Error at deleteAction: {err}")
return False
def handler(
context: dict,
inputs: dict
) -> dict:
outputs: dict = {}
actionId: str | None = None
executionId: str | None = None
try:
vcoUrl: str = context["vcoUrl"]
bearerToken: str = context["getToken"]()
actionModule: str = inputs["in_actionModule"]
actionName: str = inputs["in_actionName"]
outputType: str | None = inputs["in_outputType"]
script: str = inputs["in_script"]
version: str | None = inputs["in_version"]
runtime: str | None = inputs["in_runtime"]
action: dict = createAction(
vcoUrl,
bearerToken,
actionModule,
actionName,
outputType,
script,
version,
runtime
)
actionId = action["id"] if isinstance(action, dict) else None
if not actionId:
raise RuntimeError("Action creation returned no valid ID")
result: dict = executeAction(
vcoUrl,
bearerToken,
actionId
)
executionId = result["execution-id"] \
if isinstance(result, dict) else None
if not executionId:
raise RuntimeError("Action execution returned no valid ID")
log: dict = {}
maxAttempts = 10 # Increase for long-running or complex tests
checkInterval = 1
attemptCounter = 0
while attemptCounter < maxAttempts:
log = getActionLog(
vcoUrl,
bearerToken,
executionId
)
if log and len(log["logs"]) > 0:
break
time.sleep(checkInterval)
attemptCounter += 1
if attemptCounter >= maxAttempts:
print("Warning: Logging timeout, "
"logs might be incomplete or missing")
outputs["status"] = "done"
outputs["error"] = None
outputs["createAction"] = action
outputs["executeAction"] = result
outputs["logAction"] = log
except Exception as err:
outputs["status"] = "incomplete"
outputs["error"] = str(err)
outputs["createAction"] = None
outputs["executeAction"] = None
outputs["logAction"] = None
finally:
if actionId:
deleted: bool = deleteAction(
vcoUrl,
bearerToken,
actionId
)
outputs["deleteAction"] = deleted
else:
outputs["deleteAction"] = None
return outputs
if __name__ == "__main__":
# Call the sequence from here, if you are not in the Orchestrator
# createAction()
# executeAction()
# getActionLog()
# deleteAction()
pass
|
