@@ -1,39 +1,74 @@ | |||
# Copyright (c) 2021, Frappe Technologies and contributors | |||
# License: MIT. See LICENSE | |||
""" This is a virtual doctype controller for test/demo purposes. | |||
- It uses a JSON file on disk as "backend". | |||
- All docs are stored in "docs" key of JSON file. | |||
""" | |||
import json | |||
import os | |||
from typing import Any, TypedDict | |||
import frappe | |||
from frappe.model.document import Document | |||
DATA_FILE = "data_file.json" | |||
class Data(TypedDict): | |||
docs: dict[str, dict[str, Any]] | |||
def get_current_data() -> Data: | |||
"""Read data from disk""" | |||
if not os.path.exists(DATA_FILE): | |||
return {"docs": {}} | |||
with open(DATA_FILE) as f: | |||
return json.load(f) | |||
def update_data(data: Data) -> None: | |||
"""Flush updated data to disk""" | |||
with open(DATA_FILE, "w+") as data_file: | |||
json.dump(data, data_file) | |||
class test(Document): | |||
def db_insert(self, *args, **kwargs): | |||
d = self.get_valid_dict(convert_dates_to_str=True) | |||
with open("data_file.json", "w+") as read_file: | |||
json.dump(d, read_file) | |||
data = get_current_data() | |||
data["docs"][d.name] = d | |||
update_data(data) | |||
def load_from_db(self): | |||
with open("data_file.json") as read_file: | |||
d = json.load(read_file) | |||
super(Document, self).__init__(d) | |||
data = get_current_data() | |||
d = data["docs"].get(self.name) | |||
super(Document, self).__init__(d) | |||
def db_update(self, *args, **kwargs): | |||
d = self.get_valid_dict(convert_dates_to_str=True) | |||
with open("data_file.json", "w+") as read_file: | |||
json.dump(d, read_file) | |||
# For this example insert and update are same operation, | |||
# it might be different for you | |||
self.db_insert(*args, **kwargs) | |||
def delete(self): | |||
data = get_current_data() | |||
data["docs"].pop(self.name, None) | |||
update_data(data) | |||
@staticmethod | |||
def get_list(args): | |||
with open("data_file.json") as read_file: | |||
return [frappe._dict(json.load(read_file))] | |||
data = get_current_data() | |||
return [frappe._dict(doc) for name, doc in data["docs"].items()] | |||
@staticmethod | |||
def get_count(args): | |||
return 5 | |||
data = get_current_data() | |||
return len(data["docs"]) | |||
@staticmethod | |||
def get_stats(args): | |||
# return [] | |||
with open("data_file.json") as read_file: | |||
return [json.load(read_file)] | |||
return {} |
@@ -1,8 +1,66 @@ | |||
# Copyright (c) 2021, Frappe Technologies and Contributors | |||
# License: MIT. See LICENSE | |||
# import frappe | |||
import unittest | |||
import json | |||
import os | |||
import frappe | |||
from frappe.core.doctype.test.test import DATA_FILE | |||
from frappe.core.doctype.test.test import test as VirtDocType | |||
from frappe.desk.form.save import savedocs | |||
from frappe.tests.utils import FrappeTestCase | |||
class Testtest(unittest.TestCase): | |||
pass | |||
class Testtest(FrappeTestCase): | |||
def tearDown(self): | |||
if os.path.exists(DATA_FILE): | |||
os.remove(DATA_FILE) | |||
def test_insert_update_and_load_from_desk(self): | |||
"""Insert, update, reload and assert changes""" | |||
frappe.response.docs = [] | |||
doc = json.dumps( | |||
{ | |||
"docstatus": 0, | |||
"doctype": "test", | |||
"name": "new-test-1", | |||
"__islocal": 1, | |||
"__unsaved": 1, | |||
"owner": "Administrator", | |||
"test": "Original Data", | |||
} | |||
) | |||
savedocs(doc, "Save") | |||
docname = frappe.response.docs[0]["name"] | |||
doc = frappe.get_doc("test", docname) | |||
doc.test = "New Data" | |||
savedocs(doc.as_json(), "Save") | |||
doc.reload() | |||
self.assertEqual(doc.test, "New Data") | |||
def test_multiple_doc_insert_and_get_list(self): | |||
doc1 = frappe.get_doc(doctype="test", test="first").insert() | |||
doc2 = frappe.get_doc(doctype="test", test="second").insert() | |||
docs = {doc1.name, doc2.name} | |||
doc2.reload() | |||
doc1.reload() | |||
updated_docs = {doc1.name, doc2.name} | |||
self.assertEqual(docs, updated_docs) | |||
listed_docs = {d.name for d in VirtDocType.get_list({})} | |||
self.assertEqual(docs, listed_docs) | |||
def test_get_count(self): | |||
args = {"doctype": "test", "filters": [], "fields": []} | |||
self.assertIsInstance(VirtDocType.get_count(args), int) | |||
def test_delete_doc(self): | |||
doc = frappe.get_doc(doctype="test", test="data").insert() | |||
doc.delete() | |||
listed_docs = {d.name for d in VirtDocType.get_list({})} | |||
self.assertNotIn(doc.name, listed_docs) |
@@ -1,5 +1,7 @@ | |||
from typing import Protocol | |||
import frappe | |||
class VirtualDoctype(Protocol): | |||
"""This class documents requirements that must be met by a doctype controller to function as virtual doctype | |||
@@ -15,7 +17,7 @@ class VirtualDoctype(Protocol): | |||
# ============ class/static methods ============ | |||
@staticmethod | |||
def get_list(args): | |||
def get_list(args) -> list[frappe._dict]: | |||
"""Similar to reportview.get_list""" | |||
... | |||
@@ -41,6 +43,10 @@ class VirtualDoctype(Protocol): | |||
This is responsible for updatinng __dict__ of class with all the fields on doctype.""" | |||
... | |||
def db_update(self, *args, **kwargs): | |||
def db_update(self, *args, **kwargs) -> None: | |||
"""Serialize the `Document` object and update existing document in backend.""" | |||
... | |||
def delete(self, *args, **kwargs) -> None: | |||
"""Delete the current document from backend""" | |||
... |