|
|
@@ -1,178 +1,170 @@ |
|
|
|
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors |
|
|
|
# MIT License. See license.txt |
|
|
|
from __future__ import unicode_literals |
|
|
|
|
|
|
|
import unittest, frappe, os |
|
|
|
from frappe.core.doctype.user.user import generate_keys |
|
|
|
from frappe.frappeclient import FrappeClient, FrappeException |
|
|
|
from frappe.utils.data import get_url |
|
|
|
import unittest |
|
|
|
from random import choice |
|
|
|
|
|
|
|
import requests |
|
|
|
import base64 |
|
|
|
|
|
|
|
class TestAPI(unittest.TestCase): |
|
|
|
def test_insert_many(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabNote` where title in ('Sing','a','song','of','sixpence')") |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
server.insert_many([ |
|
|
|
{"doctype": "Note", "public": True, "title": "Sing"}, |
|
|
|
{"doctype": "Note", "public": True, "title": "a"}, |
|
|
|
{"doctype": "Note", "public": True, "title": "song"}, |
|
|
|
{"doctype": "Note", "public": True, "title": "of"}, |
|
|
|
{"doctype": "Note", "public": True, "title": "sixpence"}, |
|
|
|
]) |
|
|
|
|
|
|
|
self.assertTrue(frappe.db.get_value('Note', {'title': 'Sing'})) |
|
|
|
self.assertTrue(frappe.db.get_value('Note', {'title': 'a'})) |
|
|
|
self.assertTrue(frappe.db.get_value('Note', {'title': 'song'})) |
|
|
|
self.assertTrue(frappe.db.get_value('Note', {'title': 'of'})) |
|
|
|
self.assertTrue(frappe.db.get_value('Note', {'title': 'sixpence'})) |
|
|
|
|
|
|
|
def test_create_doc(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabNote` where title = 'test_create'") |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
server.insert({"doctype": "Note", "public": True, "title": "test_create"}) |
|
|
|
|
|
|
|
self.assertTrue(frappe.db.get_value('Note', {'title': 'test_create'})) |
|
|
|
|
|
|
|
def test_list_docs(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
doc_list = server.get_list("Note") |
|
|
|
|
|
|
|
self.assertTrue(len(doc_list)) |
|
|
|
|
|
|
|
def test_get_doc(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabNote` where title = 'get_this'") |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
server.insert_many([ |
|
|
|
{"doctype": "Note", "public": True, "title": "get_this"}, |
|
|
|
]) |
|
|
|
doc = server.get_doc("Note", "get_this") |
|
|
|
self.assertTrue(doc) |
|
|
|
|
|
|
|
def test_get_value(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabNote` where title = 'get_value'") |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
test_content = "test get value" |
|
|
|
from semantic_version import Version |
|
|
|
|
|
|
|
server.insert_many([ |
|
|
|
{"doctype": "Note", "public": True, "title": "get_value", "content": test_content}, |
|
|
|
]) |
|
|
|
self.assertEqual(server.get_value("Note", "content", {"title": "get_value"}).get('content'), test_content) |
|
|
|
name = server.get_value("Note", "name", {"title": "get_value"}).get('name') |
|
|
|
import frappe |
|
|
|
from frappe.utils import get_site_url |
|
|
|
|
|
|
|
# test by name |
|
|
|
self.assertEqual(server.get_value("Note", "content", name).get('content'), test_content) |
|
|
|
|
|
|
|
self.assertRaises(FrappeException, server.get_value, "Note", "(select (password) from(__Auth) order by name desc limit 1)", {"title": "get_value"}) |
|
|
|
|
|
|
|
def test_get_single(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
server.set_value('Website Settings', 'Website Settings', 'title_prefix', 'test-prefix') |
|
|
|
self.assertEqual(server.get_value('Website Settings', 'title_prefix', 'Website Settings').get('title_prefix'), 'test-prefix') |
|
|
|
self.assertEqual(server.get_value('Website Settings', 'title_prefix').get('title_prefix'), 'test-prefix') |
|
|
|
frappe.db.set_value('Website Settings', None, 'title_prefix', '') |
|
|
|
|
|
|
|
def test_update_doc(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabNote` where title in ('Sing','sing')") |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
server.insert({"doctype":"Note", "public": True, "title": "Sing"}) |
|
|
|
doc = server.get_doc("Note", 'Sing') |
|
|
|
changed_title = "sing" |
|
|
|
doc["title"] = changed_title |
|
|
|
doc = server.update(doc) |
|
|
|
self.assertTrue(doc["title"] == changed_title) |
|
|
|
|
|
|
|
def test_update_child_doc(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabContact` where first_name = 'George' and last_name = 'Steevens'") |
|
|
|
frappe.db.sql("delete from `tabContact` where first_name = 'William' and last_name = 'Shakespeare'") |
|
|
|
frappe.db.sql("delete from `tabCommunication` where reference_doctype = 'Event'") |
|
|
|
frappe.db.sql("delete from `tabCommunication Link` where link_doctype = 'Contact'") |
|
|
|
frappe.db.sql("delete from `tabEvent` where subject = 'Sing a song of sixpence'") |
|
|
|
frappe.db.sql("delete from `tabEvent Participants` where reference_doctype = 'Contact'") |
|
|
|
def maintain_state(f): |
|
|
|
def wrapper(*args, **kwargs): |
|
|
|
frappe.db.rollback() |
|
|
|
r = f(*args, **kwargs) |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
# create multiple contacts |
|
|
|
server.insert_many([ |
|
|
|
{"doctype": "Contact", "first_name": "George", "last_name": "Steevens"}, |
|
|
|
{"doctype": "Contact", "first_name": "William", "last_name": "Shakespeare"} |
|
|
|
]) |
|
|
|
|
|
|
|
# create an event with one of the created contacts |
|
|
|
event = server.insert({ |
|
|
|
"doctype": "Event", |
|
|
|
"subject": "Sing a song of sixpence", |
|
|
|
"event_participants": [{ |
|
|
|
"reference_doctype": "Contact", |
|
|
|
"reference_docname": "George Steevens" |
|
|
|
}] |
|
|
|
}) |
|
|
|
|
|
|
|
# update the event's contact to the second contact |
|
|
|
server.update({ |
|
|
|
"doctype": "Event Participants", |
|
|
|
"name": event.get("event_participants")[0].get("name"), |
|
|
|
"reference_docname": "William Shakespeare" |
|
|
|
}) |
|
|
|
|
|
|
|
# the change should run the parent document's validations and |
|
|
|
# create a Communication record with the new contact |
|
|
|
self.assertTrue(frappe.db.exists("Communication Link", {"link_name": "William Shakespeare"})) |
|
|
|
|
|
|
|
def test_delete_doc(self): |
|
|
|
server = FrappeClient(get_url(), "Administrator", "admin", verify=False) |
|
|
|
frappe.db.sql("delete from `tabNote` where title = 'delete'") |
|
|
|
frappe.db.commit() |
|
|
|
|
|
|
|
server.insert_many([ |
|
|
|
{"doctype": "Note", "public": True, "title": "delete"}, |
|
|
|
]) |
|
|
|
server.delete("Note", "delete") |
|
|
|
|
|
|
|
self.assertFalse(frappe.db.get_value('Note', {'title': 'delete'})) |
|
|
|
|
|
|
|
def test_auth_via_api_key_secret(self): |
|
|
|
# generate API key and API secret for administrator |
|
|
|
keys = generate_keys("Administrator") |
|
|
|
frappe.db.commit() |
|
|
|
generated_secret = frappe.utils.password.get_decrypted_password( |
|
|
|
"User", "Administrator", fieldname='api_secret' |
|
|
|
return r |
|
|
|
|
|
|
|
return wrapper |
|
|
|
|
|
|
|
|
|
|
|
class TestResourceAPI(unittest.TestCase): |
|
|
|
SITE_URL = get_site_url(frappe.local.site) |
|
|
|
RESOURCE_URL = f"{SITE_URL}/api/resource" |
|
|
|
DOCTYPE = "ToDo" |
|
|
|
GENERATED_DOCUMENTS = [] |
|
|
|
|
|
|
|
@classmethod |
|
|
|
@maintain_state |
|
|
|
def setUpClass(self): |
|
|
|
for _ in range(10): |
|
|
|
doc = frappe.get_doc( |
|
|
|
{"doctype": "ToDo", "description": frappe.mock("paragraph")} |
|
|
|
).insert() |
|
|
|
self.GENERATED_DOCUMENTS.append(doc.name) |
|
|
|
|
|
|
|
@classmethod |
|
|
|
@maintain_state |
|
|
|
def tearDownClass(self): |
|
|
|
for name in self.GENERATED_DOCUMENTS: |
|
|
|
frappe.delete_doc_if_exists(self.DOCTYPE, name) |
|
|
|
|
|
|
|
@property |
|
|
|
def sid(self): |
|
|
|
if not getattr(self, "_sid", None): |
|
|
|
self._sid = requests.post( |
|
|
|
f"{self.SITE_URL}/api/method/login", |
|
|
|
data={ |
|
|
|
"usr": "Administrator", |
|
|
|
"pwd": frappe.conf.admin_password or "admin", |
|
|
|
}, |
|
|
|
).cookies.get("sid") |
|
|
|
|
|
|
|
return self._sid |
|
|
|
|
|
|
|
def get(self, path, params=""): |
|
|
|
return requests.get(f"{self.RESOURCE_URL}/{path}?sid={self.sid}{params}") |
|
|
|
|
|
|
|
def post(self, path, data): |
|
|
|
return requests.post( |
|
|
|
f"{self.RESOURCE_URL}/{path}?sid={self.sid}", data=frappe.as_json(data) |
|
|
|
) |
|
|
|
|
|
|
|
api_key = frappe.db.get_value("User", "Administrator", "api_key") |
|
|
|
header = {"Authorization": "token {}:{}".format(api_key, generated_secret)} |
|
|
|
res = requests.post(get_url() + "/api/method/frappe.auth.get_logged_user", headers=header) |
|
|
|
|
|
|
|
self.assertEqual(res.status_code, 200) |
|
|
|
self.assertEqual("Administrator", res.json()["message"]) |
|
|
|
self.assertEqual(keys['api_secret'], generated_secret) |
|
|
|
|
|
|
|
header = {"Authorization": "Basic {}".format(base64.b64encode(frappe.safe_encode("{}:{}".format(api_key, generated_secret))).decode())} |
|
|
|
res = requests.post(get_url() + "/api/method/frappe.auth.get_logged_user", headers=header) |
|
|
|
self.assertEqual(res.status_code, 200) |
|
|
|
self.assertEqual("Administrator", res.json()["message"]) |
|
|
|
|
|
|
|
# Valid api key, invalid api secret |
|
|
|
api_secret = "ksk&93nxoe3os" |
|
|
|
header = {"Authorization": "token {}:{}".format(api_key, api_secret)} |
|
|
|
res = requests.post(get_url() + "/api/method/frappe.auth.get_logged_user", headers=header) |
|
|
|
self.assertEqual(res.status_code, 403) |
|
|
|
|
|
|
|
def put(self, path, data): |
|
|
|
return requests.put( |
|
|
|
f"{self.RESOURCE_URL}/{path}?sid={self.sid}", data=frappe.as_json(data) |
|
|
|
) |
|
|
|
|
|
|
|
# random api key and api secret |
|
|
|
api_key = "@3djdk3kld" |
|
|
|
api_secret = "ksk&93nxoe3os" |
|
|
|
header = {"Authorization": "token {}:{}".format(api_key, api_secret)} |
|
|
|
res = requests.post(get_url() + "/api/method/frappe.auth.get_logged_user", headers=header) |
|
|
|
self.assertEqual(res.status_code, 401) |
|
|
|
def delete(self, path): |
|
|
|
return requests.delete(f"{self.RESOURCE_URL}/{path}?sid={self.sid}") |
|
|
|
|
|
|
|
def test_unauthorized_call(self): |
|
|
|
# test 1: fetch documents without auth |
|
|
|
response = requests.get(f"{self.RESOURCE_URL}/{self.DOCTYPE}") |
|
|
|
self.assertEqual(response.status_code, 403) |
|
|
|
|
|
|
|
def test_get_list(self): |
|
|
|
# test 2: fetch documents without params |
|
|
|
response = self.get(self.DOCTYPE) |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertIsInstance(response.json(), dict) |
|
|
|
self.assertIn("data", response.json()) |
|
|
|
|
|
|
|
def test_get_list_limit(self): |
|
|
|
# test 3: fetch data with limit |
|
|
|
response = self.get(self.DOCTYPE, "&limit=2") |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertEqual(len(response.json()["data"]), 2) |
|
|
|
|
|
|
|
def test_get_list_dict(self): |
|
|
|
# test 4: fetch response as (not) dict |
|
|
|
response = self.get(self.DOCTYPE, "&as_dict=True") |
|
|
|
json = frappe._dict(response.json()) |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertIsInstance(json.data, list) |
|
|
|
self.assertIsInstance(json.data[0], dict) |
|
|
|
|
|
|
|
response = self.get(self.DOCTYPE, "&as_dict=False") |
|
|
|
json = frappe._dict(response.json()) |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertIsInstance(json.data, list) |
|
|
|
self.assertIsInstance(json.data[0], list) |
|
|
|
|
|
|
|
def test_get_list_debug(self): |
|
|
|
# test 5: fetch response with debug |
|
|
|
response = self.get(self.DOCTYPE, "&debug=true") |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertIn("exc", response.json()) |
|
|
|
self.assertIsInstance(response.json()["exc"], str) |
|
|
|
self.assertIsInstance(eval(response.json()["exc"]), list) |
|
|
|
|
|
|
|
def test_get_list_fields(self): |
|
|
|
# test 6: fetch response with fields |
|
|
|
response = self.get(self.DOCTYPE, r'&fields=["description"]') |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
json = frappe._dict(response.json()) |
|
|
|
self.assertIn("description", json.data[0]) |
|
|
|
|
|
|
|
def test_create_document(self): |
|
|
|
# test 7: POST method on /api/resource to create doc |
|
|
|
data = {"description": frappe.mock("paragraph")} |
|
|
|
response = self.post(self.DOCTYPE, data) |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
docname = response.json()["data"]["name"] |
|
|
|
self.assertIsInstance(docname, str) |
|
|
|
self.GENERATED_DOCUMENTS.append(docname) |
|
|
|
|
|
|
|
def test_update_document(self): |
|
|
|
# test 8: PUT method on /api/resource to update doc |
|
|
|
generated_desc = frappe.mock("paragraph") |
|
|
|
data = {"description": generated_desc} |
|
|
|
random_doc = choice(self.GENERATED_DOCUMENTS) |
|
|
|
desc_before_update = frappe.db.get_value(self.DOCTYPE, random_doc, "description") |
|
|
|
|
|
|
|
response = self.put(f"{self.DOCTYPE}/{random_doc}", data=data) |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertNotEqual(response.json()["data"]["description"], desc_before_update) |
|
|
|
self.assertEqual(response.json()["data"]["description"], generated_desc) |
|
|
|
|
|
|
|
def test_delete_document(self): |
|
|
|
# test 9: DELETE method on /api/resource |
|
|
|
doc_to_delete = choice(self.GENERATED_DOCUMENTS) |
|
|
|
response = self.delete(f"{self.DOCTYPE}/{doc_to_delete}") |
|
|
|
self.assertEqual(response.status_code, 202) |
|
|
|
self.assertDictEqual(response.json(), {"message": "ok"}) |
|
|
|
|
|
|
|
non_existent_doc = frappe.generate_hash(length=12) |
|
|
|
response = self.delete(f"{self.DOCTYPE}/{non_existent_doc}") |
|
|
|
self.assertEqual(response.status_code, 404) |
|
|
|
self.assertDictEqual(response.json(), {}) |
|
|
|
|
|
|
|
|
|
|
|
class TestMethodAPI(unittest.TestCase): |
|
|
|
METHOD_URL = f"{get_site_url(frappe.local.site)}/api/method" |
|
|
|
|
|
|
|
def test_version(self): |
|
|
|
# test 1: test for /api/method/version |
|
|
|
response = requests.get(f"{self.METHOD_URL}/version") |
|
|
|
json = frappe._dict(response.json()) |
|
|
|
|
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertIsInstance(json, dict) |
|
|
|
self.assertIsInstance(json.message, str) |
|
|
|
self.assertEqual(Version(json.message), Version(frappe.__version__)) |
|
|
|
|
|
|
|
def test_ping(self): |
|
|
|
# test 2: test for /api/method/ping |
|
|
|
response = requests.get(f"{self.METHOD_URL}/ping") |
|
|
|
self.assertEqual(response.status_code, 200) |
|
|
|
self.assertIsInstance(response.json(), dict) |
|
|
|
self.assertEqual(response.json()['message'], "pong") |