test: Split API v2 tests
Duplication here seems better than weird and hard to understand parameterized version.
This commit is contained in:
Ankush Menat
parent
76b4d209c7
commit
a46389a7e4
2 changed files with 298 additions and 202 deletions
|
|
@ -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):
|
||||
|
|
|
|||
297
frappe/tests/test_api_v2.py
Normal file
297
frappe/tests/test_api_v2.py
Normal file
|
|
@ -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)
|
||||
Loading…
Add table
Reference in a new issue