diff --git a/frappe/tests/test_api.py b/frappe/tests/test_api.py index 3825121834..9b14b308ed 100644 --- a/frappe/tests/test_api.py +++ b/frappe/tests/test_api.py @@ -13,7 +13,6 @@ from filetype import guess_mime from werkzeug.test import TestResponse import frappe -from frappe.core.doctype.user.user import get_timezones from frappe.installer import update_site_config from frappe.tests.utils import FrappeTestCase, patch_hooks from frappe.utils import cint, get_test_client, get_url @@ -75,22 +74,6 @@ class ThreadWithReturnValue(Thread): return self._return -def parameterize(*api_versions): - def decorator(f): - def wrapper(*args, **kwargs): - original_version = args[0].version - try: - for api_version in api_versions: - args[0].version = api_version - f(*args, **kwargs) - finally: - args[0].version = original_version - - return wrapper - - return decorator - - resource_key = { "": "resource", "v1": "resource", @@ -164,13 +147,11 @@ class TestResourceAPI(FrappeAPITestCase): frappe.delete_doc_if_exists(cls.DOCTYPE, name) frappe.db.commit() - @parameterize("", "v1", "v2") def test_unauthorized_call(self): # test 1: fetch documents without auth response = requests.get(self.resource_path(self.DOCTYPE)) self.assertEqual(response.status_code, 403) - @parameterize("", "v1", "v2") def test_get_list(self): # test 2: fetch documents without params response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid}) @@ -178,14 +159,12 @@ class TestResourceAPI(FrappeAPITestCase): self.assertIsInstance(response.json, dict) self.assertIn("data", response.json) - @parameterize("", "v1", "v2") def test_get_list_limit(self): # test 3: fetch data with limit response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "limit": 2}) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.json["data"]), 2) - @parameterize("", "v1", "v2") def test_get_list_dict(self): # test 4: fetch response as (not) dict response = self.get(self.resource_path(self.DOCTYPE), {"sid": self.sid, "as_dict": True}) @@ -200,7 +179,6 @@ class TestResourceAPI(FrappeAPITestCase): self.assertIsInstance(json.data, list) self.assertIsInstance(json.data[0], list) - @parameterize("", "v1") def test_get_list_debug(self): # test 5: fetch response with debug with suppress_stdout(): @@ -210,7 +188,6 @@ class TestResourceAPI(FrappeAPITestCase): self.assertIsInstance(response.json["exc"], str) self.assertIsInstance(eval(response.json["exc"]), list) - @parameterize("", "v1", "v2") def test_get_list_fields(self): # test 6: fetch response with fields response = self.get( @@ -220,7 +197,6 @@ class TestResourceAPI(FrappeAPITestCase): json = frappe._dict(response.json) self.assertIn("description", json.data[0]) - @parameterize("", "v1", "v2") def test_create_document(self): data = {"description": frappe.mock("paragraph"), "sid": self.sid} response = self.post(self.resource_path(self.DOCTYPE), data) @@ -229,7 +205,6 @@ class TestResourceAPI(FrappeAPITestCase): self.assertIsInstance(docname, str) self.GENERATED_DOCUMENTS.append(docname) - @parameterize("", "v1") def test_update_document(self): generated_desc = frappe.mock("paragraph") data = {"description": generated_desc, "sid": self.sid} @@ -242,7 +217,6 @@ class TestResourceAPI(FrappeAPITestCase): response = self.get(self.resource_path(self.DOCTYPE, random_doc)) self.assertEqual(response.json["data"]["description"], generated_desc) - @parameterize("", "v1", "v2") def test_delete_document(self): doc_to_delete = choice(self.GENERATED_DOCUMENTS) response = self.delete(self.resource_path(self.DOCTYPE, doc_to_delete)) @@ -253,7 +227,6 @@ class TestResourceAPI(FrappeAPITestCase): self.assertEqual(response.status_code, 404) self.GENERATED_DOCUMENTS.remove(doc_to_delete) - @parameterize("", "v1") def test_run_doc_method(self): # test 10: Run whitelisted method on doc via /api/resource # status_code is 403 if no other tests are run before this - it's not logged in @@ -278,7 +251,7 @@ class TestResourceAPI(FrappeAPITestCase): self.assertIsInstance(data[0], dict) -class TestMethodAPIV1(FrappeAPITestCase): +class TestMethodAPI(FrappeAPITestCase): def test_ping(self): # test 2: test for /api/method/ping response = self.get(self.method_path("ping")) @@ -343,169 +316,6 @@ class TestMethodAPIV1(FrappeAPITestCase): self.assertIn("Traceback", response.json["exc"]) -class TestDocumentAPIV2(TestResourceAPI): - version = "v2" - - def setUp(self) -> None: - self.post(self.method_path("login"), {"sid": self.sid}) - return super().setUp() - - def test_execute_doc_method(self): - response = self.get(self.resource_path("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_path(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_path(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_path(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_path("login"), {"sid": self.sid}) - return super().setUp() - - def test_ping(self): - response = self.get(self.method_path("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_path("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_path("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_path("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_path("User", "get_all_roles"), {"sid": self.sid}) - self.assertIn("Blogger", shorthand_response.json["data"]) - - expanded_response = self.get( - self.method_path("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_path("logout"), {"sid": self.sid}) - response = self.get(self.method_path("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_path("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_path("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_path(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_path(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_path(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"]) - - -class TestDocTypeAPIV2(FrappeAPITestCase): - version = "v2" - - def setUp(self) -> None: - self.post(self.method_path("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""" @@ -518,14 +328,12 @@ class TestReadOnlyMode(FrappeAPITestCase): # XXX: this has potential to crumble rest of the test suite. update_site_config("maintenance_mode", 1) - @parameterize("", "v1", "v2") def test_reads(self): response = self.get(self.resource_path("ToDo"), {"sid": self.sid}) self.assertEqual(response.status_code, 200) self.assertIsInstance(response.json, dict) self.assertIsInstance(response.json["data"], list) - @parameterize("", "v1") def test_blocked_writes(self): with suppress_stdout(): response = self.post( @@ -534,15 +342,6 @@ class TestReadOnlyMode(FrappeAPITestCase): self.assertEqual(response.status_code, 503) self.assertEqual(response.json["exc_type"], "InReadOnlyMode") - @parameterize("v2") - def test_blocked_writes_v2(self): - with suppress_stdout(): - response = self.post( - self.resource_path("ToDo"), {"description": frappe.mock("paragraph"), "sid": self.sid} - ) - self.assertEqual(response.status_code, 503) - self.assertEqual(response.json["errors"][0]["type"], "InReadOnlyMode") - class TestWSGIApp(FrappeAPITestCase): def test_request_hooks(self): diff --git a/frappe/tests/test_api_v2.py b/frappe/tests/test_api_v2.py new file mode 100644 index 0000000000..ba25b8b05d --- /dev/null +++ b/frappe/tests/test_api_v2.py @@ -0,0 +1,297 @@ +from random import choice + +import requests + +import frappe +from frappe.installer import update_site_config +from frappe.tests.test_api import FrappeAPITestCase, suppress_stdout + +authorization_token = None + + +resource_key = { + "": "resource", + "v1": "resource", + "v2": "document", +} + + +class TestResourceAPIV2(FrappeAPITestCase): + version = "v2" + DOCTYPE = "ToDo" + GENERATED_DOCUMENTS = [] + + @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): + 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_path(self.DOCTYPE)) + self.assertEqual(response.status_code, 403) + + def test_get_list(self): + # test 2: fetch documents without params + response = self.get(self.resource_path(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_path(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_path(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_path(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_path(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_path(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_delete_document(self): + doc_to_delete = choice(self.GENERATED_DOCUMENTS) + response = self.delete(self.resource_path(self.DOCTYPE, doc_to_delete)) + self.assertEqual(response.status_code, 202) + self.assertDictEqual(response.json, {"data": "ok"}) + + response = self.get(self.resource_path(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_path("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_path(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_path(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_path(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_path("login"), {"sid": self.sid}) + return super().setUp() + + def test_ping(self): + response = self.get(self.method_path("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_path("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_path("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_path("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_path("User", "get_all_roles"), {"sid": self.sid}) + self.assertIn("Blogger", shorthand_response.json["data"]) + + expanded_response = self.get( + self.method_path("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_path("logout"), {"sid": self.sid}) + response = self.get(self.method_path("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_path("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_path("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_path(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_path(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_path(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"]) + + +class TestDocTypeAPIV2(FrappeAPITestCase): + version = "v2" + + def setUp(self) -> None: + self.post(self.method_path("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_path("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_path("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)