diff --git a/frappe/api.py b/frappe/api.py index 9039ae0e5f..6427cbfbd8 100644 --- a/frappe/api.py +++ b/frappe/api.py @@ -11,6 +11,7 @@ import frappe.client import frappe.handler from frappe import _ from frappe.utils.response import build_response +from frappe.utils.data import sbool def handle(): @@ -108,25 +109,40 @@ def handle(): elif doctype: if frappe.local.request.method == "GET": - if frappe.local.form_dict.get('fields'): - frappe.local.form_dict['fields'] = json.loads(frappe.local.form_dict['fields']) - frappe.local.form_dict.setdefault('limit_page_length', 20) - frappe.local.response.update({ - "data": frappe.call( - frappe.client.get_list, - doctype, - **frappe.local.form_dict - ) - }) + # set fields for frappe.get_list + if frappe.local.form_dict.get("fields"): + frappe.local.form_dict["fields"] = json.loads(frappe.local.form_dict["fields"]) + + # set limit of records for frappe.get_list + frappe.local.form_dict.setdefault( + "limit_page_length", + frappe.local.form_dict.limit or frappe.local.form_dict.limit_page_length or 20, + ) + + # convert strings to native types - only as_dict and debug accept bool + for param in ["as_dict", "debug"]: + param_val = frappe.local.form_dict.get(param) + if param_val is not None: + frappe.local.form_dict[param] = sbool(param_val) + + # evaluate frappe.get_list + data = frappe.call(frappe.client.get_list, doctype, **frappe.local.form_dict) + + # set frappe.get_list result to response + frappe.local.response.update({"data": data}) if frappe.local.request.method == "POST": + # fetch data from from dict data = get_request_form_data() - data.update({ - "doctype": doctype - }) - frappe.local.response.update({ - "data": frappe.get_doc(data).insert().as_dict() - }) + data.update({"doctype": doctype}) + + # insert document from request data + doc = frappe.get_doc(data).insert() + + # set response data + frappe.local.response.update({"data": doc.as_dict()}) + + # commit for POST requests frappe.db.commit() else: raise frappe.DoesNotExistError diff --git a/frappe/tests/test_api.py b/frappe/tests/test_api.py index 6453062877..7e77aab779 100644 --- a/frappe/tests/test_api.py +++ b/frappe/tests/test_api.py @@ -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") diff --git a/frappe/tests/test_frappe_client.py b/frappe/tests/test_frappe_client.py new file mode 100644 index 0000000000..e1cdbb6ccd --- /dev/null +++ b/frappe/tests/test_frappe_client.py @@ -0,0 +1,177 @@ +# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors +# MIT License. See license.txt + +import unittest, frappe +from frappe.core.doctype.user.user import generate_keys +from frappe.frappeclient import FrappeClient, FrappeException +from frappe.utils.data import get_url + +import requests +import base64 + +class TestFrappeClient(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" + + 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') + + # 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'") + 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' + ) + + 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) + + + # 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) diff --git a/frappe/utils/data.py b/frappe/utils/data.py index 9cbac2a570..a9af30ab2c 100644 --- a/frappe/utils/data.py +++ b/frappe/utils/data.py @@ -622,6 +622,26 @@ def ceil(s): def cstr(s, encoding='utf-8'): return frappe.as_unicode(s, encoding) +def sbool(x): + """Converts str object to Boolean if possible. + Example: + "true" becomes True + "1" becomes True + "{}" remains "{}" + + Args: + x (str): String to be converted to Bool + + Returns: + object: Returns Boolean or type(x) + """ + from distutils.util import strtobool + + try: + return bool(strtobool(x)) + except Exception: + return x + def rounded(num, precision=0): """round method for round halfs to nearest even algorithm aka banker's rounding - compatible with python3""" precision = cint(precision)