import typing from random import choice import requests import frappe from frappe.installer import update_site_config from frappe.tests.test_api import FrappeAPITestCase, suppress_stdout from frappe.tests.utils import toggle_test_mode authorization_token = None resource_key = { "": "resource", "v1": "resource", "v2": "document", } class TestResourceAPIV2(FrappeAPITestCase): version = "v2" DOCTYPE = "ToDo" GENERATED_DOCUMENTS: typing.ClassVar[list] = [] @classmethod def setUpClass(cls): super().setUpClass() for _ in range(20): doc = frappe.get_doc({"doctype": "ToDo", "description": frappe.mock("paragraph")}).insert() cls.GENERATED_DOCUMENTS = [] cls.GENERATED_DOCUMENTS.append(doc.name) frappe.db.commit() @classmethod def tearDownClass(cls): frappe.db.commit() for name in cls.GENERATED_DOCUMENTS: frappe.delete_doc_if_exists(cls.DOCTYPE, name) frappe.db.commit() def test_unauthorized_call(self): # test 1: fetch documents without auth response = requests.get(self.resource(self.DOCTYPE)) self.assertEqual(response.status_code, 403) def test_get_list(self): # test 2: fetch documents without params response = self.get(self.resource(self.DOCTYPE), {"sid": self.sid}) 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.resource(self.DOCTYPE), {"sid": self.sid, "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.resource(self.DOCTYPE), {"sid": self.sid, "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.resource(self.DOCTYPE), {"sid": self.sid, "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_fields(self): # test 6: fetch response with fields response = self.get(self.resource(self.DOCTYPE), {"sid": self.sid, "fields": '["description"]'}) self.assertEqual(response.status_code, 200) json = frappe._dict(response.json) self.assertIn("description", json.data[0]) def test_create_document(self): data = {"description": frappe.mock("paragraph"), "sid": self.sid} response = self.post(self.resource(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_copy_document(self): doc = frappe.get_doc(self.DOCTYPE, self.GENERATED_DOCUMENTS[0]) # disabled temporarily to assert that `docstatus` is not copied outside of tests toggle_test_mode(False) try: response = self.get(self.resource(self.DOCTYPE, doc.name, "copy")) finally: toggle_test_mode(True) self.assertEqual(response.status_code, 200) data = response.json["data"] self.assertEqual(data["doctype"], self.DOCTYPE) self.assertEqual(data["description"], doc.description) self.assertEqual(data["status"], doc.status) self.assertEqual(data["priority"], doc.priority) self.assertNotIn("name", data) self.assertNotIn("creation", data) self.assertNotIn("modified", data) self.assertNotIn("modified_by", data) self.assertNotIn("owner", data) self.assertNotIn("docstatus", data) def test_delete_document(self): doc_to_delete = choice(self.GENERATED_DOCUMENTS) response = self.delete(self.resource(self.DOCTYPE, doc_to_delete), data={"sid": self.sid}) self.assertEqual(response.status_code, 202) self.assertDictEqual(response.json, {"data": "ok"}) response = self.get(self.resource(self.DOCTYPE, doc_to_delete)) self.assertEqual(response.status_code, 404) self.GENERATED_DOCUMENTS.remove(doc_to_delete) def test_execute_doc_method(self): response = self.get(self.resource("Website Theme", "Standard", "method", "get_apps")) self.assertEqual(response.json["data"][0]["name"], "frappe") def test_update_document(self): generated_desc = frappe.mock("paragraph") data = {"description": generated_desc, "sid": self.sid} random_doc = choice(self.GENERATED_DOCUMENTS) response = self.patch(self.resource(self.DOCTYPE, random_doc), data=data) self.assertEqual(response.status_code, 200) self.assertEqual(response.json["data"]["description"], generated_desc) response = self.get(self.resource(self.DOCTYPE, random_doc)) self.assertEqual(response.json["data"]["description"], generated_desc) def test_delete_document_non_existing(self): non_existent_doc = frappe.generate_hash(length=12) with suppress_stdout(): response = self.delete(self.resource(self.DOCTYPE, non_existent_doc)) self.assertEqual(response.status_code, 404) self.assertEqual(response.json["errors"][0]["type"], "DoesNotExistError") # 404s dont return exceptions self.assertFalse(response.json["errors"][0].get("exception")) class TestMethodAPIV2(FrappeAPITestCase): version = "v2" def setUp(self) -> None: self.post(self.method("login"), {"sid": self.sid}) return super().setUp() def test_ping(self): response = self.get(self.method("ping")) self.assertEqual(response.status_code, 200) self.assertIsInstance(response.json, dict) self.assertEqual(response.json["data"], "pong") def test_get_user_info(self): response = self.get(self.method("frappe.realtime.get_user_info")) self.assertEqual(response.status_code, 200) self.assertIsInstance(response.json, dict) self.assertIn(response.json.get("data").get("user"), ("Administrator", "Guest")) def test_auth_cycle(self): global authorization_token generate_admin_keys() user = frappe.get_doc("User", "Administrator") api_key, api_secret = user.api_key, user.get_password("api_secret") authorization_token = f"{api_key}:{api_secret}" response = self.get(self.method("frappe.auth.get_logged_user")) self.assertEqual(response.status_code, 200) self.assertEqual(response.json["data"], "Administrator") authorization_token = None def test_404s(self): response = self.get(self.get_path("rest"), {"sid": self.sid}) self.assertEqual(response.status_code, 404) response = self.get(self.resource("User", "NonExistent@s.com"), {"sid": self.sid}) self.assertEqual(response.status_code, 404) def test_shorthand_controller_methods(self): shorthand_response = self.get(self.method("User", "get_all_roles"), {"sid": self.sid}) self.assertIn("Website Manager", shorthand_response.json["data"]) expanded_response = self.get( self.method("frappe.core.doctype.user.user.get_all_roles"), {"sid": self.sid} ) self.assertEqual(expanded_response.data, shorthand_response.data) def test_logout(self): self.post(self.method("logout"), {"sid": self.sid}) response = self.get(self.method("ping")) self.assertFalse(response.request.cookies["sid"]) def test_run_doc_method_in_memory(self): dns = frappe.get_doc("Document Naming Settings") # Check that simple API can be called. response = self.get( self.method("run_doc_method"), { "sid": self.sid, "document": dns.as_dict(), "method": "get_transactions_and_prefixes", }, ) self.assertTrue(response.json["data"]) self.assertGreaterEqual(len(response.json["docs"]), 1) # Call with known and unknown arguments, only known should get passed response = self.get( self.method("run_doc_method"), { "sid": self.sid, "document": dns.as_dict(), "method": "get_options", "kwargs": {"doctype": "Webhook", "unknown": "what"}, }, ) self.assertEqual(response.status_code, 200) def test_logs(self): method = "frappe.tests.test_api.test" expected_message = "Failed v2" response = self.get(self.method(method), {"sid": self.sid, "message": expected_message}).json self.assertIsInstance(response["messages"], list) self.assertEqual(response["messages"][0]["message"], expected_message) # Cause handled failured with suppress_stdout(): response = self.get( self.method(method), {"sid": self.sid, "message": expected_message, "fail": True} ).json self.assertIsInstance(response["errors"], list) self.assertEqual(response["errors"][0]["message"], expected_message) self.assertEqual(response["errors"][0]["type"], "ValidationError") self.assertIn("Traceback", response["errors"][0]["exception"]) # Cause handled failured with suppress_stdout(): response = self.get( self.method(method), {"sid": self.sid, "message": expected_message, "fail": True, "handled": False}, ).json self.assertIsInstance(response["errors"], list) self.assertEqual(response["errors"][0]["type"], "ZeroDivisionError") self.assertIn("Traceback", response["errors"][0]["exception"]) def test_add_comment(self): comment_txt = frappe.generate_hash() response = self.post( self.resource("User", "Administrator", "method", "add_comment"), {"text": comment_txt} ).json self.assertEqual(response["data"]["content"], comment_txt) class TestDocTypeAPIV2(FrappeAPITestCase): version = "v2" def setUp(self) -> None: self.post(self.method("login"), {"sid": self.sid}) return super().setUp() def test_meta(self): response = self.get(self.doctype_path("ToDo", "meta")) self.assertEqual(response.status_code, 200) self.assertEqual(response.json["data"]["name"], "ToDo") def test_count(self): response = self.get(self.doctype_path("ToDo", "count")) self.assertIsInstance(response.json["data"], int) class TestReadOnlyMode(FrappeAPITestCase): """During migration if read only mode can be enabled. Test if reads work well and writes are blocked""" version = "v2" @classmethod def setUpClass(cls): super().setUpClass() update_site_config("allow_reads_during_maintenance", 1) cls.addClassCleanup(update_site_config, "maintenance_mode", 0) update_site_config("maintenance_mode", 1) def test_reads(self): response = self.get(self.resource("ToDo"), {"sid": self.sid}) self.assertEqual(response.status_code, 200) self.assertIsInstance(response.json, dict) self.assertIsInstance(response.json["data"], list) def test_blocked_writes_v2(self): with suppress_stdout(): response = self.post( self.resource("ToDo"), {"description": frappe.mock("paragraph"), "sid": self.sid} ) self.assertEqual(response.status_code, 503) self.assertEqual(response.json["errors"][0]["type"], "InReadOnlyMode") def generate_admin_keys(): from frappe.core.doctype.user.user import generate_keys generate_keys("Administrator") frappe.db.commit() @frappe.whitelist() def test(*, fail=False, handled=True, message="Failed"): if fail: if handled: frappe.throw(message) else: 1 / 0 else: frappe.msgprint(message)