"""RPA platform connectors (app/crud/rpa_providers.py, app/routers/rpa.py, the rpa_run / rpa_job_status Task nodes in app/crud/tasks.py), with every platform's HTTP API stubbed by an httpx MockTransport (no network).""" import json import httpx import pytest from app.crud import rpa_providers as rp ORCH = "https://cloud.uipath.com/acme/DefaultTenant/orchestrator_" class FakeUiPath: """Just enough of Automation Cloud: token, folder lookup, releases, StartJobs, and a job that is Running once and then finishes.""" def __init__(self, final_state="Successful"): self.final_state = final_state self.polls = 0 self.started_with: dict | None = None self.folder_headers: list[str | None] = [] def __call__(self, request: httpx.Request) -> httpx.Response: url = str(request.url) if url == "https://cloud.uipath.com/identity_/connect/token": form = dict(x.split("=", 1) for x in request.content.decode().split("&")) if form.get("client_secret") != "s3cret": return httpx.Response(400, json={"error": "invalid_client"}) return httpx.Response(200, json={"access_token": "tok", "expires_in": 3600}) assert request.headers["authorization"] == "Bearer tok" path = request.url.path.removeprefix("/acme/DefaultTenant/orchestrator_") if path == "/odata/Folders": return httpx.Response(200, json={"value": [{"Id": 42, "FullyQualifiedName": "Finance/AP"}]}) self.folder_headers.append(request.headers.get("x-uipath-organizationunitid")) if path == "/odata/Releases": return httpx.Response(200, json={"value": [{"Key": "11111111-2222-3333-4444-555555555555", "Name": "InvoiceBot", "ProcessKey": "InvoiceBot", "Description": "Posts invoices"}]}) if path == "/odata/Jobs/UiPath.Server.Configuration.OData.StartJobs": self.started_with = json.loads(request.content)["startInfo"] return httpx.Response(201, json={"value": [{"Id": 9001, "State": "Pending"}]}) if path == "/odata/Jobs(9001)": self.polls += 1 if self.polls == 1: return httpx.Response(200, json={"Id": 9001, "State": "Running"}) outputs = json.dumps({"InvoiceNo": "INV-77", "Lines": 3}) if self.final_state == "Successful" else None return httpx.Response(200, json={"Id": 9001, "State": self.final_state, "OutputArguments": outputs, "Info": "done" if outputs else "Selector not found"}) if path == "/odata/Queues/UiPathODataSvc.AddQueueItem": item = json.loads(request.content)["itemData"] assert item["Name"] == "Invoices" and item["Reference"] == "R-1" and item["SpecificContent"] == {"vendor": "ACME"} return httpx.Response(201, json={"Id": 555, "Status": "New"}) return httpx.Response(404, json={"message": f"unexpected {path}"}) @pytest.fixture def mock_http(monkeypatch): """Route every adapter request to the given handler; polling doesn't sleep.""" def install(handler): monkeypatch.setattr(rp, "_client", lambda timeout=rp.TIMEOUT: httpx.AsyncClient(transport=httpx.MockTransport(handler))) return handler async def no_sleep(_): return None monkeypatch.setattr(rp, "_sleep", no_sleep) rp._TOKENS.clear() rp.UiPath._folders.clear() return install UIPATH_CONFIG = {"provider": "uipath", "orchestrator_url": ORCH, "client_id": "app", "client_secret": "s3cret", "folder": "Finance/AP"} async def _rpa_connector(client, auth_headers, config=UIPATH_CONFIG): resp = await client.post("/connectors", headers=auth_headers, json={"name": "Test RPA", "type": "rpa", "configuration": config, "status": "enabled"}) assert resp.status_code == 201, resp.text return resp.json()["id"] async def _run(client, auth_headers, nodes, context): task = await client.post("/tasks", headers=auth_headers, json={"name": "RPA test", "nodes": nodes}) assert task.status_code == 201, task.text resp = await client.post(f"/tasks/{task.json()['id']}/run", headers=auth_headers, json={"context": context, "allow_draft": True}) assert resp.status_code == 200, resp.text return resp.json() def _graph(rpa_node: dict) -> list[dict]: return [ {"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "bot"}]}, {"key": "bot", "type": "rpa_run", "label": "Run bot", **rpa_node}, ] async def test_providers_are_listed_with_capabilities(client, auth_headers): resp = await client.get("/rpa/providers", headers=auth_headers) assert resp.status_code == 200 by_id = {p["id"]: p for p in resp.json()} assert set(by_id) == {"uipath", "automation_anywhere", "power_automate", "blue_prism", "custom"} assert "queues" in by_id["uipath"]["capabilities"] and "status" not in by_id["blue_prism"]["capabilities"] assert {f["key"]: f for f in by_id["uipath"]["fields"]}["client_secret"]["secret"] is True async def test_uipath_job_runs_to_completion_through_a_task(client, auth_headers, mock_http): fake = mock_http(FakeUiPath()) connector_id = await _rpa_connector(client, auth_headers) result = await _run( client, auth_headers, _graph({ "connector_id": connector_id, "rpa_process": "InvoiceBot", "input_params": [{"name": "VendorId", "type": "string", "source_field": "vendor"}], "output_params": [{"name": "job_id", "type": "string"}, {"name": "status", "type": "string"}, {"name": "InvoiceNo", "type": "string"}], }), {"vendor": "V-9"}, ) assert result["ok"], result["trace"] assert {k: result["context"][k] for k in ("job_id", "status", "InvoiceNo")} == {"job_id": "9001", "status": "succeeded", "InvoiceNo": "INV-77"} assert fake.started_with["ReleaseKey"] == "11111111-2222-3333-4444-555555555555" assert json.loads(fake.started_with["InputArguments"]) == {"VendorId": "V-9"} assert set(fake.folder_headers) == {"42"} # every call scoped to the Finance/AP folder assert "job 9001 succeeded" in result["trace"][-1]["detail"] async def test_failed_job_fails_the_node_unless_told_to_continue(client, auth_headers, mock_http): mock_http(FakeUiPath(final_state="Faulted")) connector_id = await _rpa_connector(client, auth_headers) failing = await _run(client, auth_headers, _graph({"connector_id": connector_id, "rpa_process": "InvoiceBot"}), {}) assert not failing["ok"] and "failed" in failing["trace"][-1]["detail"] and "Selector not found" in failing["trace"][-1]["detail"] mock_http(FakeUiPath(final_state="Faulted")) tolerant = await _run( client, auth_headers, _graph({"connector_id": connector_id, "rpa_process": "InvoiceBot", "rpa_fail_on_error": False, "output_params": [{"name": "status", "type": "string"}]}), {}, ) assert tolerant["ok"] and tolerant["context"]["status"] == "failed" async def test_uipath_queue_item_without_waiting(client, auth_headers, mock_http): mock_http(FakeUiPath()) connector_id = await _rpa_connector(client, auth_headers) result = await _run( client, auth_headers, _graph({ "connector_id": connector_id, "rpa_action": "add_queue_item", "rpa_queue": "Invoices", "rpa_wait": False, "input_params": [{"name": "vendor", "type": "string", "source_field": "vendor"}, {"name": "reference", "type": "string", "source_field": "ref"}], "output_params": [{"name": "job_id", "type": "string"}], }), {"vendor": "ACME", "ref": "R-1"}, ) assert result["ok"], result["trace"] assert result["context"]["job_id"] == "queue:555" async def test_processes_picker_and_test_button(client, auth_headers, mock_http): mock_http(FakeUiPath()) connector_id = await _rpa_connector(client, auth_headers) procs = await client.get(f"/rpa/connectors/{connector_id}/processes", headers=auth_headers) assert procs.status_code == 200 and procs.json() == [{"id": "InvoiceBot", "name": "InvoiceBot", "description": "Posts invoices"}] test = await client.post(f"/connectors/{connector_id}/test", headers=auth_headers) assert test.json()["success"] is True and "accepted" in test.json()["detail"] bad = await _rpa_connector(client, auth_headers, {**UIPATH_CONFIG, "client_secret": "wrong"}) rp._TOKENS.clear() test = await client.post(f"/connectors/{bad}/test", headers=auth_headers) assert test.json()["success"] is False and "rejected" in test.json()["detail"] async def test_power_automate_async_flow_is_polled(mock_http): calls = [] def handler(request: httpx.Request) -> httpx.Response: calls.append((request.method, str(request.url))) if request.method == "POST": assert json.loads(request.content) == {"amount": 10} return httpx.Response(202, headers={"location": "https://flow.test/runs/1"}) return httpx.Response(202) if len(calls) < 3 else httpx.Response(200, json={"approved": True}) mock_http(handler) config = {"provider": "power_automate", "flows": "Approve | https://prod.logic.azure.com/workflows/x/invoke?sig=abc\n\nhttps://other.logic.azure.com/y?sig=d"} assert rp.power_automate_flows(config) == {"Approve": "https://prod.logic.azure.com/workflows/x/invoke?sig=abc", "Flow 2": "https://other.logic.azure.com/y?sig=d"} job = await rp.run_process(config, action="start_job", process="Approve", inputs={"amount": 10}, timeout=60) assert job.status == "succeeded" and job.outputs == {"approved": True} and len(calls) == 3 async def test_automation_anywhere_deploy_and_outputs(mock_http): def handler(request: httpx.Request) -> httpx.Response: body = json.loads(request.content or b"{}") if request.url.path == "/v2/authentication": return httpx.Response(200, json={"token": "aa-tok"}) assert request.headers["x-authorization"] == "aa-tok" if request.url.path == "/v3/automations/deploy": assert body["botId"] == 17 and body["unattendedRequest"]["runAsUserIds"] == [5] assert body["botInput"] == {"count": {"type": "NUMBER", "number": "3"}, "name": {"type": "STRING", "string": "x"}} return httpx.Response(200, json={"deploymentId": "dep-1"}) if request.url.path == "/v3/activity/list": return httpx.Response(200, json={"list": [{"status": "COMPLETED", "botOutVariables": {"values": {"total": {"type": "NUMBER", "number": "12"}}}}]}) return httpx.Response(404) mock_http(handler) config = {"provider": "automation_anywhere", "control_room_url": "https://cr.test", "username": "bot", "api_key": "k", "run_as_user_ids": "5"} job = await rp.run_process(config, action="start_job", process="17", inputs={"count": 3, "name": "x"}, timeout=60) assert (job.job_id, job.status, job.outputs) == ("dep-1", "succeeded", {"total": 12}) async def test_queues_are_refused_where_unsupported(): with pytest.raises(rp.RpaError, match="queues"): await rp.run_process({"provider": "blue_prism", "runtime_url": "http://bp:8181"}, action="add_queue_item", process="", inputs={}, queue="Q")