vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 4

test: add tests for webhook additions

Browse source
This commit is contained in:
Rohan Bansal 2019-08-27 16:24:40 +05:30
parent 2e86ed58ec
commit ef31f44a0b
2 changed files with 110 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
frappe/integrations/doctype/webhook/test_webhook.py
View file

@ -3,19 +3,102 @@
# See license.txt
from __future__ import unicode_literals
import frappe
import unittest
import frappe
from frappe.integrations.doctype.webhook.webhook import get_webhook_headers, get_webhook_data
class TestWebhook(unittest.TestCase):
def test_validate_docevents(self):
def test_validate_doc_events(self):
"Test creating a submit-related webhook for a non-submittable DocType"
doc = frappe.new_doc("Webhook")
doc.webhook_doctype = "User"
doc.webhook_docevent = "on_submit"
doc.request_url = "https://httpbin.org/post"
self.assertRaises(frappe.ValidationError, doc.save)
def test_validate_request_url(self):
"Test validation for the webhook request URL"
doc = frappe.new_doc("Webhook")
doc.webhook_doctype = "User"
doc.webhook_docevent = "after_insert"
doc.request_url = "httpbin.org?post"
self.assertRaises(frappe.ValidationError, doc.save)
def test_validate_headers(self):
"Test validation for request headers"
doc = frappe.new_doc("Webhook")
doc.webhook_doctype = "User"
doc.webhook_docevent = "after_insert"
doc.request_url = "https://httpbin.org/post"
# test incomplete headers
doc.webhook_headers = [{
"key": "Content-Type"
}]
doc.save()
headers = get_webhook_headers(doc=None, webhook=doc)
self.assertEqual(headers, None)
# test complete headers
doc.webhook_headers = [{
"key": "Content-Type",
"value": "application/json"
}]
doc.save()
headers = get_webhook_headers(doc=None, webhook=doc)
self.assertEqual(headers, {"Content-Type": "application/json"})
def test_validate_request_body_form(self):
"Test validation of Form URL-Encoded request body"
doc = frappe.new_doc("Webhook")
doc.webhook_doctype = "User"
doc.webhook_docevent = "after_insert"
doc.request_url = "https://httpbin.org/post"
doc.request_structure = "Form URL-Encoded"
doc.set("webhook_data", [{
"fieldname": "name",
"key": "name"
}])
doc.webhook_json = """{
"name": "Test User"
}"""
doc.save()
self.assertEqual(doc.webhook_json, None)
user = frappe.new_doc("User")
user.first_name = frappe.mock("name")
user.email = frappe.mock("email")
user.save()
data = get_webhook_data(user, webhook=doc)
self.assertEqual(data, {"name": user.name})
def test_validate_request_body_json(self):
"Test validation of JSON request body"
doc = frappe.new_doc("Webhook")
doc.webhook_doctype = "User"
doc.webhook_docevent = "after_insert"
doc.request_url = "https://httpbin.org/post"
doc.request_structure = "JSON"
doc.set("webhook_data", [{
"fieldname": "name",
"key": "name"
}])
doc.webhook_json = """{
"name": "{{ doc.name }}"
}"""
doc.save()
self.assertEqual(doc.webhook_data, [])
user = frappe.new_doc("User")
user.first_name = frappe.mock("name")
user.email = frappe.mock("email")
user.save()
data = get_webhook_data(user, webhook=doc)
self.assertEqual(data, {"name": user.name})

41
frappe/integrations/doctype/webhook/webhook.py
View file

@ -77,22 +77,8 @@ def get_context(doc):
def enqueue_webhook(doc, webhook):
webhook = frappe.get_doc("Webhook", webhook.get("name"))
headers = {}
data = {}
if webhook.webhook_headers:
for h in webhook.webhook_headers:
if h.get("key") and h.get("value"):
headers[h.get("key")] = h.get("value")
if webhook.webhook_data:
for w in webhook.webhook_data:
value = doc.get(w.fieldname)
if isinstance(value, datetime.datetime):
value = frappe.utils.get_datetime_str(value)
data[w.key] = value
elif webhook.webhook_json:
data = frappe.render_template(webhook.webhook_json, get_context(doc))
data = json.loads(data)
headers = get_webhook_headers(doc, webhook)
data = get_webhook_data(doc, webhook)
for i in range(3):
try:
@ -107,3 +93,26 @@ def enqueue_webhook(doc, webhook):
continue
else:
raise e
def get_webhook_headers(doc, webhook):
headers = {}
if webhook.webhook_headers:
for h in webhook.webhook_headers:
if h.get("key") and h.get("value"):
headers[h.get("key")] = h.get("value")
return headers
def get_webhook_data(doc, webhook):
data = {}
if webhook.webhook_data:
for w in webhook.webhook_data:
value = doc.get(w.fieldname)
if isinstance(value, datetime.datetime):
value = frappe.utils.get_datetime_str(value)
data[w.key] = value
elif webhook.webhook_json:
data = frappe.render_template(webhook.webhook_json, get_context(doc))
data = json.loads(data)
return data