add test for robocorp client

.github/workflows/main.yml

@@ -47,5 +47,6 @@ jobs:
         working-directory: ./external-worker/
       - run: pip install -e ".[testing]"
         working-directory: ./robocorp/
+      - run: cp -a external-worker/flowable/external_worker_client robocorp/flowable/ # we need to copy this module over, since it's only searching locally for modules
       - run: python -m unittest discover
         working-directory: ./robocorp/

robocorp/README.md

@@ -1,4 +1,4 @@
-# Flowable Robocorp Client
+from typing import Dict# Flowable Robocorp Client
 
 [License:
 ![license](https://img.shields.io/hexpm/l/plug.svg)](https://github.com/flowable/flowable-external-client-python/blob/main/LICENSE)
@@ -31,7 +31,7 @@ The following example `get-weather-forecast.py` can be used as a Robocorp action
 from robocorp.actions import action
 
 @action
-def get_weather_forecast(city: str, days: int, scale: str = "celsius") -> str:
+def get_weather_forecast(city: str, days: int, scale: str = "celsius") -> dict[str, str]:
     """
     Returns weather conditions forecast for a given city.
 

robocorp/flowable/.gitignore

@@ -0,0 +1 @@
+external_worker_client

robocorp/setup.py

@@ -17,6 +17,6 @@
     license='Apache License, Version 2.0',
     install_requires=['flowable.external-worker-client>=1.0.1rc1', 'robocorp-actions>=0.2.1', 'robocorp-tasks>=3.1.1', 'robocorp-truststore>=0.9.1'],
     extras_require={
-        'testing': ['pytest', 'vcrpy']
+        'testing': ['pytest', 'vcrpy', 'flowable.external-worker-client>=1.0.1rc1']
     },
 )

robocorp/tests/basic_test.py

@@ -0,0 +1,20 @@
+import unittest
+from pathlib import Path
+
+from requests.auth import HTTPBasicAuth
+
+from tests.bpmn_utils import deploy_process, get_process_definition_id, delete_deployment
+
+base_url = "http://localhost:8090/flowable-work"
+auth = HTTPBasicAuth("admin", "test")
+
+
+class BasicTest(unittest.TestCase):
+
+    def deploy_process(self, fileToDeploy='externalWorkerProcess.bpmn'):
+        self._process_deployment_id = deploy_process(base_url, auth, str(Path(__file__).parent.absolute()) + '/fixtures/processes/' + fileToDeploy)
+        self._process_definition_id = get_process_definition_id(base_url, auth, self._process_deployment_id)
+
+    def remove_deployment(self):
+        if hasattr(self, "_process_deployment_id"):
+            delete_deployment(base_url, auth, self._process_deployment_id)

robocorp/tests/bpmn_utils.py

@@ -0,0 +1,55 @@
+import requests
+from requests.auth import AuthBase
+
+
+def deploy_process(base_url: str, auth: AuthBase, file: str) -> str:
+    files = {'file': open(file, 'rb')}
+    deployment_result = requests.post(base_url + '/process-api/repository/deployments', auth=auth, files=files)
+    assert deployment_result.status_code == 201
+    return deployment_result.json().get("id")
+
+
+def delete_deployment(base_url: str, auth: AuthBase, process_deployment_id: str) -> None:
+    delete_deployment_result = requests.delete(base_url + "/process-api/repository/deployments/" + process_deployment_id, auth=auth)
+    assert delete_deployment_result.status_code == 204
+
+
+def get_process_definition_id(base_url: str, auth: AuthBase, process_deployment_id: str) -> None:
+    definition_search = requests.get(base_url + "/process-api/repository/process-definitions?deploymentId=" + process_deployment_id, auth=auth)
+    assert definition_search.status_code == 200
+    data = definition_search.json().get("data")
+    assert len(data) == 1
+    return data[0].get("id")
+
+
+def start_process(base_url: str, auth: AuthBase, process_definition_id: str, variables: list[object] = []) -> str:
+    start_result = requests.post(
+        base_url + "/process-api/runtime/process-instances",
+        auth=auth,
+        json={"processDefinitionId": process_definition_id, "variables": variables},
+        headers={"Content-Type": "application/json"}
+    )
+    assert start_result.status_code == 201
+    return start_result.json().get("id")
+
+
+def terminate_process(base_url: str, auth: AuthBase, process_instance_id: str) -> None:
+    delete_result = requests.delete(base_url + "/process-api/runtime/process-instances/" + process_instance_id, auth=auth)
+    assert delete_result.status_code == 204 or delete_result.status_code == 404
+
+
+def get_process_variable(base_url: str, auth: AuthBase, process_instance_id: str, variable_name: str) -> dict | None:
+    variable_result = requests.get(base_url + "/process-api/history/historic-variable-instances?processInstanceId=" + process_instance_id + "&variableName=" + variable_name, auth=auth)
+    assert variable_result.status_code == 200
+    data = variable_result.json().get("data")
+    if len(data) == 1:
+        return data[0].get("variable")
+    else:
+        return None
+
+
+def executed_activity_ids(base_url: str, auth: AuthBase, process_instance_id: str) -> list[str]:
+    activity_instances = requests.get(base_url + "/process-api/history/historic-activity-instances?processInstanceId=" + process_instance_id, auth=auth)
+    assert activity_instances.status_code == 200
+    data = activity_instances.json().get("data")
+    return sorted(list(map(lambda o: o.get("activityId"), data)))