From 6252440f6567135c6deb0700c22934ff30690003 Mon Sep 17 00:00:00 2001 From: ManyaGirdhar Date: Sat, 27 Jun 2026 19:06:57 +0530 Subject: [PATCH 1/5] chore: add clm as required app --- documenso_integration/hooks.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/documenso_integration/hooks.py b/documenso_integration/hooks.py index 057c71d..7e7c661 100644 --- a/documenso_integration/hooks.py +++ b/documenso_integration/hooks.py @@ -8,7 +8,7 @@ # Apps # ------------------ -# required_apps = [] +required_apps = ["clm"] # Each item in the list will be shown as an app in the apps page # add_to_apps_screen = [ From 72a13f851ac0ba282257bdf84cebe1b48b39f7cd Mon Sep 17 00:00:00 2001 From: ManyaGirdhar Date: Sat, 27 Jun 2026 19:08:54 +0530 Subject: [PATCH 2/5] sec: load api config from Documenso Settings and check permissions --- documenso_integration/api.py | 156 ++++++++++++++++------------------- 1 file changed, 69 insertions(+), 87 deletions(-) diff --git a/documenso_integration/api.py b/documenso_integration/api.py index 4c3dc58..f0319b9 100644 --- a/documenso_integration/api.py +++ b/documenso_integration/api.py @@ -1,56 +1,63 @@ -import os import requests import frappe -# from dotenv import load_dotenv +from frappe import _ from frappe.utils.pdf import get_pdf -# load_dotenv() - -DOCUMENSO_API_KEY = os.getenv("DOCUMENSO_API_KEY") -DOCUMENSO_BASE_URL = os.getenv("DOCUMENSO_BASE_URL") +def get_settings(): + return frappe.get_single("Documenso Settings") + +def get_headers(): + settings = get_settings() + api_key = settings.get_password("api_key") + if not api_key: + frappe.throw(_("Documenso API Key is not configured in Documenso Settings.")) + return { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json" + } -headers = { - "Authorization": f"Bearer {DOCUMENSO_API_KEY}", - "Content-Type": "application/json" -} +def get_base_url(): + settings = get_settings() + return settings.base_url or "https://documenso.com/api/v1/documents" -@frappe.whitelist() +@frappe.whitelist(methods=["POST"]) def sign_contract(contract_name): + frappe.has_permission("Contract", "write", doc=contract_name, throw=True) + create_response = create_documenso_record(contract_name) if "error" in create_response: - frappe.throw(f"Documenso record creation failed: {create_response['error']}") + frappe.throw(_("Documenso record creation failed: {0}").format(create_response['error'])) return upload_response = upload_contract_to_documenso(contract_name) - if "error" in upload_response: - frappe.throw(f"Documenso Upload Error: {upload_response['error']}") + frappe.throw(_("Documenso Upload Error: {0}").format(upload_response['error'])) field_response = create_field_in_documenso(contract_name) if "error" in field_response: - frappe.throw(f"Documenso Field Creation Error: {field_response['error']}") + frappe.throw(_("Documenso Field Creation Error: {0}").format(field_response['error'])) send_response = send_contract_for_signature(contract_name) if "error" in send_response: - frappe.throw(f"Documenso Send Error: {send_response['error']}") + frappe.throw(_("Documenso Send Error: {0}").format(send_response['error'])) -@frappe.whitelist() +@frappe.whitelist(methods=["POST"]) def create_documenso_record(contract_name): + frappe.has_permission("Contract", "write", doc=contract_name, throw=True) + contract = frappe.get_doc("Contract", contract_name) - # Retrieve all signees from the child table - signees = contract.signee # Assuming 'signee' is the child table fieldname + signees = contract.signee if not signees: - return {"error": "No signees found in the contract."} + return {"error": _("No signees found in the contract.")} - # Construct the recipients list recipients = [] for idx, signee in enumerate(signees): recipients.append({ "email": signee.email, - "name": signee.name or "Signee", # Replace with appropriate field if available + "name": signee.name or _("Signee"), "role": "SIGNER", - "signingOrder": idx # Adjust signing order as needed + "signingOrder": idx }) payload = { @@ -58,65 +65,54 @@ def create_documenso_record(contract_name): "externalId": contract.name, "recipients": recipients, "meta": { - "signingOrder": "SEQUENTIAL" # Use "PARALLEL" if order doesn't matter + "signingOrder": "SEQUENTIAL" }, "authOptions": {}, "formValues": {} } - - response = requests.post(DOCUMENSO_BASE_URL, json=payload, headers=headers) + response = requests.post(get_base_url(), json=payload, headers=get_headers(), timeout=10) if response.status_code == 200: doc_data = response.json() - print(doc_data) - # frappe.db.set_value("Contract", contract.name, "document_id", doc_data.get("documentId")) - # frappe.db.set_value("Contract", contract.name, "upload_url", doc_data.get("uploadUrl")) if "recipients" in doc_data and doc_data["recipients"]: - print("\nRecipients:", doc_data["recipients"], "\n") recipient_ids = [str(r.get("recipientId")) for r in doc_data["recipients"] if r.get("recipientId")] joined_ids = ",".join(recipient_ids) - # frappe.db.set_value("Contract", contract.name, "recipient_id", joined_ids) + signing_url = [str(r.get("signingUrl")) for r in doc_data["recipients"] if r.get("signingUrl")] joined_urls = ",".join(signing_url) - # frappe.db.set_value("Contract", contract.name, "signing_url", joined_urls) -# for api information - # Step 3: Create API Information record api_info = frappe.get_doc({ "doctype": "API Information", "document_id": doc_data.get("documentId"), "upload_url": doc_data.get("uploadUrl"), "recipient_id": joined_ids, "signing_url": joined_urls, - # "download_url": doc_data.get("downloadUrl") # if available }) api_info.insert(ignore_permissions=True) - # Step 4: Link API Information to Contract contract.db_set("documenso_id", api_info.name) - frappe.db.commit() - - return doc_data else: frappe.log_error(f"Documenso Error: {response.text}", "Documenso API") return {"error": response.text} -@frappe.whitelist() +@frappe.whitelist(methods=["POST"]) def upload_contract_to_documenso(contract_name): try: + frappe.has_permission("Contract", "write", doc=contract_name, throw=True) + contract = frappe.get_doc("Contract", contract_name) pdf_file = get_pdf(frappe.get_print("Contract", contract_name, print_format="Contract Final Print")) api_info = frappe.get_doc("API Information", contract.documenso_id) - # upload_url = contract.upload_url - upload_url = api_info.upload_url # Use the upload URL from API Information + + upload_url = api_info.upload_url if not upload_url: - frappe.throw("Upload URL is missing. Ensure the document is created in Documenso.") + frappe.throw(_("Upload URL is missing. Ensure the document is created in Documenso.")) files = {'file': (f"{contract_name}.pdf", pdf_file, "application/pdf")} - response = requests.put(upload_url, files=files) + response = requests.put(upload_url, files=files, timeout=10) if response.status_code in [200, 201]: return {"success": True} @@ -125,29 +121,29 @@ def upload_contract_to_documenso(contract_name): except Exception as e: return {"error": str(e)} -@frappe.whitelist() +@frappe.whitelist(methods=["POST"]) def create_field_in_documenso(contract_name): + frappe.has_permission("Contract", "write", doc=contract_name, throw=True) + contract = frappe.get_doc("Contract", contract_name) - # document_id = contract.document_id api_info = frappe.get_doc("API Information", contract.documenso_id) document_id = api_info.document_id recipient_ids_str = api_info.recipient_id if not document_id or not recipient_ids_str: - return {"error": "Document ID or Recipient ID(s) is missing."} + return {"error": _("Document ID or Recipient ID(s) is missing.")} try: document_id = int(document_id) except ValueError: - return {"error": "Invalid Document ID format."} + return {"error": _("Invalid Document ID format.")} - # Split the recipient IDs and convert to integers try: recipient_ids = [int(rid.strip()) for rid in recipient_ids_str.split(",") if rid.strip()] except ValueError: - return {"error": "Invalid Recipient ID(s) format."} + return {"error": _("Invalid Recipient ID(s) format.")} - url = f"{DOCUMENSO_BASE_URL}/{document_id}/fields" + url = f"{get_base_url()}/{document_id}/fields" errors = [] for recipient_id in recipient_ids: @@ -162,7 +158,7 @@ def create_field_in_documenso(contract_name): "fieldMeta": {} } - response = requests.post(url, json=payload, headers=headers) + response = requests.post(url, json=payload, headers=get_headers(), timeout=10) if response.status_code not in [200, 201]: errors.append({ "recipient_id": recipient_id, @@ -170,40 +166,36 @@ def create_field_in_documenso(contract_name): }) if errors: - return {"error": "Some fields could not be created.", "details": errors} + return {"error": _("Some fields could not be created."), "details": errors} else: return {"success": True} - -@frappe.whitelist() +@frappe.whitelist(methods=["POST"]) def send_contract_for_signature(contract_name): - # Fetch contract document + frappe.has_permission("Contract", "write", doc=contract_name, throw=True) + contract = frappe.get_doc("Contract", contract_name) - # document_id = contract.document_id api_info = frappe.get_doc("API Information", contract.documenso_id) document_id = api_info.document_id if not document_id: - return {"error": "Document ID is missing."} - - url = f"{DOCUMENSO_BASE_URL}/{int(document_id)}/send" - print("\nSending to:", url, "\n") + return {"error": _("Document ID is missing.")} + url = f"{get_base_url()}/{int(document_id)}/send" payload = { "sendEmail": True, "sendCompletionEmails": True } try: - response = requests.post(url, headers=headers, json=payload) + response = requests.post(url, headers=get_headers(), json=payload, timeout=10) response_data = response.json() - print(response_data) except Exception as e: frappe.log_error( title="Documenso Response Error", message=f"Error parsing response: {e}\nRaw response: {response.text}" ) - return {"error": "Failed to parse Documenso response."} + return {"error": _("Failed to parse Documenso response.")} if response.status_code in [200, 201]: try: @@ -214,25 +206,17 @@ def send_contract_for_signature(contract_name): recipients = response_data if not recipients: - return {"error": "No recipients found in response."} + return {"error": _("No recipients found in response.")} - # Extract signing URLs corresponding to existing recipient IDs - recipient_ids = [rid.strip() for rid in api_info.recipient_id.split(",") if rid.strip()] - print(recipient_ids) - - # Update the contract document api_info.save(ignore_permissions=True) - frappe.db.commit() - - return {"message": "Document sent and signing URLs saved successfully."} + return {"message": _("Document sent and signing URLs saved successfully.")} except Exception as e: frappe.log_error( title="Error Saving Signing URLs", message=f"Exception: {e}\nResponse: {frappe.as_json(response_data)}" ) - return {"error": "Error while saving signing URLs."} - + return {"error": _("Error while saving signing URLs.")} else: frappe.log_error( title="Documenso API Error", @@ -240,40 +224,38 @@ def send_contract_for_signature(contract_name): ) return {"error": response_data} -@frappe.whitelist() +@frappe.whitelist(methods=["POST"]) def download_signed_contract(contract_name): + frappe.has_permission("Contract", "write", doc=contract_name, throw=True) + contract = frappe.get_doc("Contract", contract_name) - # document_id = contract.document_id api_info = frappe.get_doc("API Information", contract.documenso_id) document_id = api_info.document_id if not document_id: - return {"error": "Document ID is missing."} + return {"error": _("Document ID is missing.")} - url = f"{DOCUMENSO_BASE_URL}/{int(document_id)}/download" - response = requests.get(url, headers=headers) + url = f"{get_base_url()}/{int(document_id)}/download" + response = requests.get(url, headers=get_headers(), timeout=10) if response.status_code == 200: try: data = response.json() download_url = data.get("downloadUrl") - frappe.log_error("Download URL:", download_url) if not download_url: - return {"error": "Download URL not found in the response."} + return {"error": _("Download URL not found in the response.")} - # Save the download URL to the API Information doctype frappe.db.set_value("API Information", api_info.name, "download_url", download_url) - frappe.db.commit() return { "success": True, - "message": "Download URL saved successfully.", + "message": _("Download URL saved successfully."), "download_url": download_url } except Exception as e: frappe.log_error(f"Download URL Parsing Error: {e}\nResponse: {response.text}") - return {"error": "Failed to parse download URL from Documenso."} + return {"error": _("Failed to parse download URL from Documenso.")} else: frappe.log_error(f"Documenso Download URL Error: {response.text}") - return {"error": f"Failed to get download URL. Status code: {response.status_code}"} + return {"error": _("Failed to get download URL. Status code: {0}").format(response.status_code)} From 44bcce2ac5740a154291b67c9287a7d4cdd56563 Mon Sep 17 00:00:00 2001 From: ManyaGirdhar Date: Sat, 27 Jun 2026 19:09:01 +0530 Subject: [PATCH 3/5] sec: verify webhook hmac and handle state transitions --- documenso_integration/webhook.py | 143 +++++++++++++------------------ 1 file changed, 58 insertions(+), 85 deletions(-) diff --git a/documenso_integration/webhook.py b/documenso_integration/webhook.py index ab75878..973f0d2 100644 --- a/documenso_integration/webhook.py +++ b/documenso_integration/webhook.py @@ -1,125 +1,98 @@ import frappe -import os +import hmac import json from frappe import _ from documenso_integration.api import download_signed_contract - @frappe.whitelist(allow_guest=True) def incoming_webhook(): - # Get the webhook key from the environment variable - expected_key = os.getenv("DOCUMENSO_WEBHOOK_KEY") - # Get received secret key from header - received_key = frappe.request.headers.get("X-Documenso-Secret") - # Log the keys for debugging - frappe.log_error(f"Expected Key: {expected_key}\nReceived Key: {received_key}", "Documenso Webhook Auth") + settings = frappe.get_single("Documenso Settings") + expected_secret = settings.get_password("webhook_secret") + + if not expected_secret: + frappe.throw(_("Webhook secret is not configured in Documenso Settings."), frappe.PermissionError) - # Check if the keys match, otherwise throw an error - if expected_key and received_key != expected_key: - frappe.throw(_("Unauthorized request"), frappe.PermissionError) + received_signature = frappe.request.headers.get("X-Documenso-Secret") or frappe.request.headers.get("X-Documenso-Signature") + + if not received_signature or not hmac.compare_digest(received_signature, expected_secret): + frappe.throw(_("Unauthorized request signature."), frappe.PermissionError) - # Capture the webhook payload payload = frappe.request.get_json() if not payload: frappe.throw(_("Invalid or missing JSON payload"), frappe.BadRequest) - # Log the payload to verify frappe.log_error(json.dumps(payload, indent=2), "Incoming Documenso Webhook") - # Extract the necessary information from the payload event = payload.get('event') - frappe.log_error(f"Event: {event}", "Documenso Webhook Event") document_data = payload.get('payload', {}) - frappe.log_error(f"Document Data: {json.dumps(document_data, indent=2)}", "Documenso Webhook Document Data") document_id = document_data.get('id') - frappe.log_error(f"Document ID: {document_id}", "Documenso Webhook Document ID") status = document_data.get('status') - frappe.log_error(f"Status: {status}", "Documenso Webhook Status") if not document_id or not status: frappe.throw(_("Missing required data in payload")) - # Get the document to update try: - # doc = frappe.get_doc("Contract", {"document_id": document_id}) doc = frappe.get_doc("API Information", {"document_id": document_id}) - frappe.log_error(f"Document found: {doc.name}", "Documenso Webhook Document Found") + frappe.log_error(f"API Information document found: {doc.name}", "Documenso Webhook Document Found") except frappe.DoesNotExistError: - frappe.throw(_("Document not found")) + frappe.throw(_("API Information record not found for external Document ID: {0}").format(document_id)) - # # Handle different events and update the workflow state accordingly if event == "DOCUMENT_COMPLETED" and status == "COMPLETED": try: - download_signed_contract(doc.name) # This should update the download_url field + download_signed_contract(doc.name) + + contract_name = frappe.db.get_value("Contract", {"documenso_id": doc.name}, "name") + if contract_name: + contract = frappe.get_doc("Contract", contract_name) + if contract.workflow_state != "Active": + contract.db_set("workflow_state", "Active") + frappe.log_error(f"Contract {contract.name} set to 'Active' on DOCUMENT_COMPLETED", "Documenso Webhook") + + frappe.publish_realtime('workflow_state_updated', { + 'contract_name': contract.name, + 'workflow_state': "Active" + }) + else: + frappe.log_error(f"No Contract found with documenso_id (API Info name): {doc.name}", "Documenso Completed Event") except Exception as e: - frappe.log_error(f"Error downloading signed contract for {doc.name}: {str(e)}", "Documenso Download Error") + frappe.log_error(f"Error downloading/updating signed contract for {doc.name}: {str(e)}", "Documenso Download Error") - # frappe.db.set_value("Contract", doc.name, "workflow_state", "Active") - # frappe.log_error(f"Before Save - Workflow State: {doc.workflow_state}", "Debug Save") - - # frappe.publish_realtime('workflow_state_updated', { - # 'contract_name': doc.name, - # 'workflow_state': doc.workflow_state - # }) - # frappe.log_error(f"Real-time event published: {doc.name} updated to Active", "Documenso Webhook Publish Realtime") - - # try: - # download_signed_contract(doc.name) - # except Exception as e: - # frappe.log_error(f"Error downloading signed contract for {doc.name}: {str(e)}", "Documenso Download Error") - - elif event == "DOCUMENT_REJECTED": try: - if doc.document_id: - # Find the Contract using document_id - contract_name = frappe.db.get_value("Contract", {"documenso_id": doc.document_id}, "name") - - if contract_name: - contract = frappe.get_doc("Contract", contract_name) - if contract.workflow_state != "Active": - contract.workflow_state = "Active" - contract.save(ignore_permissions=True) - frappe.log_error(f"Contract {contract.name} set to 'Active' on DOCUMENT_REJECTED", "Documenso Webhook") - else: - frappe.log_error(f"No Contract found with document_id: {doc.document_id}", "Documenso Rejected Event") + contract_name = frappe.db.get_value("Contract", {"documenso_id": doc.name}, "name") + if contract_name: + contract = frappe.get_doc("Contract", contract_name) + if contract.workflow_state != "Rejected": + contract.db_set("workflow_state", "Rejected") + frappe.log_error(f"Contract {contract.name} set to 'Rejected' on DOCUMENT_REJECTED", "Documenso Webhook") + + frappe.publish_realtime('workflow_state_updated', { + 'contract_name': contract.name, + 'workflow_state': "Rejected" + }) + else: + frappe.log_error(f"No Contract found with documenso_id (API Info name): {doc.name}", "Documenso Rejected Event") except Exception as e: frappe.log_error(f"Error handling DOCUMENT_REJECTED for {doc.name}: {str(e)}", "Documenso Rejected Event Error") + return {"status": "success", "message": "Webhook processed successfully"} - # elif event == "DOCUMENT_CANCELLED" and status == "PENDING": - # doc.workflow_state = "Rejected" - # frappe.log_error(f"Document {document_id} canceled and updated to 'Canceled'", "Documenso Webhook") - - # else: - # frappe.log_error(f"Unhandled event/status:\nEvent: {event}\nStatus: {status}\nDocument ID: {document_id}", "Documenso Webhook") - # frappe.throw(_("Unhandled event or status")) - - # # Save the document with the updated workflow state - # doc.save() - - # # Log the final workflow state - # frappe.log_error(f"Document {document_id} successfully updated to state: {doc.workflow_state}", "Documenso Webhook") - - return {"status": "success", "message": "Webhook processed and document updated"} - -@frappe.whitelist(allow_guest=True) def api_information_after_save(doc, method): - # Only proceed if download_url is filled if doc.download_url: - # Update linked Contract workflow_state to "Active" - if doc.document_id: - try: - # Get the Contract doc name using document_id - contract_name = frappe.db.get_value("Contract", {"documenso_id": doc.document_id}, "name") - - if contract_name: - contract = frappe.get_doc("Contract", contract_name) - if contract.workflow_state != "Active": - contract.workflow_state = "Active" - contract.save(ignore_permissions=True) - else: - frappe.log_error(f"No Contract found with document_id: {doc.document_id}", "API Information Doc Events") + try: + contract_name = frappe.db.get_value("Contract", {"documenso_id": doc.name}, "name") + + if contract_name: + contract = frappe.get_doc("Contract", contract_name) + if contract.workflow_state != "Active": + contract.db_set("workflow_state", "Active") + + frappe.publish_realtime('workflow_state_updated', { + 'contract_name': contract.name, + 'workflow_state': "Active" + }) + else: + frappe.log_error(f"No Contract found with documenso_id (API Info name): {doc.name}", "API Information Doc Events") - except Exception as e: - frappe.log_error(f"Failed to update Contract for document_id {doc.document_id}: {str(e)}", "API Information Doc Events") + except Exception as e: + frappe.log_error(f"Failed to update Contract for API Info name {doc.name}: {str(e)}", "API Information Doc Events") From 0b25b1935d37f83d5b2c3d6e00afbf20913b63b1 Mon Sep 17 00:00:00 2001 From: ManyaGirdhar Date: Sat, 27 Jun 2026 19:09:14 +0530 Subject: [PATCH 4/5] test: verify webhook authentication and contract state changes --- .../api_information/test_api_information.py | 146 +++++++++++++++++- 1 file changed, 144 insertions(+), 2 deletions(-) diff --git a/documenso_integration/documenso_integration/doctype/api_information/test_api_information.py b/documenso_integration/documenso_integration/doctype/api_information/test_api_information.py index ee33a78..52c9f17 100644 --- a/documenso_integration/documenso_integration/doctype/api_information/test_api_information.py +++ b/documenso_integration/documenso_integration/doctype/api_information/test_api_information.py @@ -1,9 +1,151 @@ # Copyright (c) 2025, Manya and Contributors # See license.txt -# import frappe +import frappe +import hmac +import json from frappe.tests.utils import FrappeTestCase +from documenso_integration.webhook import incoming_webhook +class MockRequest: + def __init__(self, headers, json_data): + self.headers = headers + self.json_data = json_data + + def get_json(self): + return self.json_data class TestAPIInformation(FrappeTestCase): - pass + def setUp(self): + # Clean up any leftover test data first + frappe.db.delete("Contract Version", {"contract": "test-webhook-contract"}) + frappe.db.delete("Contract", {"counterparty_name": "Test Acme Corp"}) + frappe.db.delete("CounterParty", {"organization": "Test Acme Corp"}) + frappe.db.delete("Contract Content", {"contract_type": "Confidentiality & Non-Disclosure Contract"}) + frappe.db.delete("API Information", {"document_id": "test_doc_123"}) + frappe.db.delete("Singles", {"doctype": "Documenso Settings"}) + + # Create test Documenso Settings + self.settings = frappe.get_doc({ + "doctype": "Documenso Settings", + "base_url": "https://documenso.com/api/v1/documents", + "api_key": "test_api_key", + "webhook_secret": "test_webhook_secret" + }) + self.settings.insert(ignore_permissions=True) + + # Create a test Contract Content + if not frappe.db.exists("Contract Content", "Confidentiality & Non-Disclosure Contract"): + self.template = frappe.get_doc({ + "doctype": "Contract Content", + "contract_type": "Confidentiality & Non-Disclosure Contract", + "content": "

Test template content

" + }) + self.template.insert(ignore_permissions=True) + + # Create a test Counterparty + if not frappe.db.exists("CounterParty", "Test Acme Corp"): + self.counterparty = frappe.get_doc({ + "doctype": "CounterParty", + "organization": "Test Acme Corp", + "requester_email": "test_acme@example.com", + "requester_phone_no": "+919999999999", + "address": "Test Address", + "organization_website": "https://testacme.com" + }) + self.counterparty.insert(ignore_permissions=True) + + # Create a test Contract + self.contract = frappe.get_doc({ + "doctype": "Contract", + "title": "Test Webhook Contract", + "contract_type": "Confidentiality & Non-Disclosure Contract", + "counterparty_name": "Test Acme Corp", + "counterparty_email": "test_acme@example.com", + "contract_term": "Definite", + "contract_duration": 12, + "contract_effective_date": "2026-06-26", + "content": "

Test Content

", + "amount_to_receive": 1000, + "tax": 18, + "requester_name": "Test Requester", + "requester_department": "Legal" + }) + self.contract.insert(ignore_permissions=True) + self.contract.db_set("workflow_state", "Awaiting Signature") + + # Create linked API Information + self.api_info = frappe.get_doc({ + "doctype": "API Information", + "document_id": "test_doc_123", + "upload_url": "https://upload.com/test", + "recipient_id": "999", + "signing_url": "https://sign.com/999" + }) + self.api_info.insert(ignore_permissions=True) + + # Link them + self.contract.db_set("documenso_id", self.api_info.name) + + # Save original request + self.original_request = getattr(frappe.local, "request", None) + + def tearDown(self): + frappe.local.request = self.original_request + frappe.db.delete("Contract Version", {"contract": "test-webhook-contract"}) + frappe.db.delete("Contract", {"counterparty_name": "Test Acme Corp"}) + frappe.db.delete("CounterParty", {"organization": "Test Acme Corp"}) + frappe.db.delete("Contract Content", {"contract_type": "Confidentiality & Non-Disclosure Contract"}) + frappe.db.delete("API Information", {"document_id": "test_doc_123"}) + frappe.db.delete("Singles", {"doctype": "Documenso Settings"}) + + def test_webhook_unauthorized(self): + frappe.local.request = MockRequest( + headers={"X-Documenso-Secret": "wrong_secret"}, + json_data={} + ) + with self.assertRaises(frappe.PermissionError): + incoming_webhook() + + def test_webhook_document_completed(self): + frappe.local.request = MockRequest( + headers={"X-Documenso-Secret": "test_webhook_secret"}, + json_data={ + "event": "DOCUMENT_COMPLETED", + "payload": { + "id": "test_doc_123", + "status": "COMPLETED" + } + } + ) + + import documenso_integration.webhook as webhook_module + original_download = webhook_module.download_signed_contract + webhook_module.download_signed_contract = lambda name: frappe.db.set_value("API Information", name, "download_url", "https://download.com/signed") + + try: + response = incoming_webhook() + self.assertEqual(response.get("status"), "success") + + contract_state = frappe.db.get_value("Contract", self.contract.name, "workflow_state") + self.assertEqual(contract_state, "Active") + finally: + webhook_module.download_signed_contract = original_download + + def test_webhook_document_rejected(self): + frappe.local.request = MockRequest( + headers={"X-Documenso-Secret": "test_webhook_secret"}, + json_data={ + "event": "DOCUMENT_REJECTED", + "payload": { + "id": "test_doc_123", + "status": "REJECTED" + } + } + ) + + response = incoming_webhook() + self.assertEqual(response.get("status"), "success") + + contract_state = frappe.db.get_value("Contract", self.contract.name, "workflow_state") + self.assertEqual(contract_state, "Rejected") From e023565cad1eb9fdb7e0ba6c2dc0165664724fe3 Mon Sep 17 00:00:00 2001 From: ManyaGirdhar Date: Sun, 28 Jun 2026 00:13:49 +0530 Subject: [PATCH 5/5] ci: fix python 3.10 compatibility and add clm dependency --- .github/workflows/ci.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a419b48..c0ca533 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -82,16 +82,18 @@ jobs: - name: Setup run: | pip install frappe-bench - bench init --skip-redis-config-generation --skip-assets --python "$(which python)" ~/frappe-bench + bench init --skip-redis-config-generation --skip-assets --frappe-branch version-15 --python "$(which python)" ~/frappe-bench mariadb --host 127.0.0.1 --port 3306 -u root -proot -e "SET GLOBAL character_set_server = 'utf8mb4'" mariadb --host 127.0.0.1 --port 3306 -u root -proot -e "SET GLOBAL collation_server = 'utf8mb4_unicode_ci'" - name: Install working-directory: /home/runner/frappe-bench run: | + bench get-app clm https://github.com/ManyaGirdhar/Contract-Lifecycle-Management-Software.git bench get-app documenso_integration $GITHUB_WORKSPACE bench setup requirements --dev bench new-site --db-root-password root --admin-password admin test_site + bench --site test_site install-app clm bench --site test_site install-app documenso_integration bench build env: