253 lines
8 KiB
Python
253 lines
8 KiB
Python
# Copyright (c) 2012 Web Notes Technologies Pvt Ltd (http://erpnext.com)
|
|
#
|
|
# MIT License (MIT)
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a
|
|
# copy of this software and associated documentation files (the "Software"),
|
|
# to deal in the Software without restriction, including without limitation
|
|
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
|
# and/or sell copies of the Software, and to permit persons to whom the
|
|
# Software is furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
|
|
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
|
|
# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
|
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
|
|
# CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE
|
|
# OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
|
#
|
|
|
|
# Tree (Hierarchical) Nested Set Model (nsm)
|
|
#
|
|
# To use the nested set model,
|
|
# use the following pattern
|
|
# 1. name your parent field as "parent_node" if not have a property nsm_parent_field as your field name in the document class
|
|
# 2. have a field called "old_parent" in your fields list - this identifies whether the parent has been changed
|
|
# 3. call update_nsm(doc_obj) in the on_upate method
|
|
|
|
# ------------------------------------------
|
|
|
|
import webnotes, unittest
|
|
|
|
class TestNSM(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
from webnotes.model.doc import Document
|
|
self.root = Document(fielddata={'doctype':'nsmtest', 'name':'T001', 'parent':None})
|
|
self.first_child = Document(fielddata={'doctype':'nsmtest', 'name':'C001', 'parent_node':'T001'})
|
|
self.first_sibling = Document(fielddata={'doctype':'nsmtest', 'name':'C002', 'parent_node':'T001'})
|
|
self.grand_child = Document(fielddata={'doctype':'nsmtest', 'name':'GC001', 'parent_node':'C001'})
|
|
|
|
webnotes.conn.sql("""
|
|
create table `tabnsmtest` (
|
|
name varchar(120) not null primary key,
|
|
creation datetime,
|
|
modified datetime,
|
|
modified_by varchar(40),
|
|
owner varchar(40),
|
|
docstatus int(1) default '0',
|
|
parent varchar(120),
|
|
parentfield varchar(120),
|
|
parenttype varchar(120),
|
|
idx int(8),
|
|
parent_node varchar(180),
|
|
old_parent varchar(180),
|
|
lft int,
|
|
rgt int) ENGINE=InnoDB""")
|
|
|
|
def test_root(self):
|
|
self.root.save(1)
|
|
update_nsm(self.root)
|
|
self.assertTrue(self.root.lft==1)
|
|
self.assertTrue(self.root.rgt==2)
|
|
|
|
def test_first_child(self):
|
|
self.root.save(1)
|
|
update_nsm(self.root)
|
|
|
|
self.first_child.save(1)
|
|
update_nsm(self.first_child)
|
|
|
|
self.root._loadfromdb()
|
|
|
|
self.assertTrue(self.root.lft==1)
|
|
self.assertTrue(self.first_child.lft==2)
|
|
self.assertTrue(self.first_child.rgt==3)
|
|
self.assertTrue(self.root.rgt==4)
|
|
|
|
def test_sibling(self):
|
|
self.test_first_child()
|
|
|
|
self.first_sibling.save(1)
|
|
update_nsm(self.first_sibling)
|
|
|
|
self.root._loadfromdb()
|
|
self.first_child._loadfromdb()
|
|
|
|
self.assertTrue(self.root.lft==1)
|
|
self.assertTrue(self.first_child.lft==2)
|
|
self.assertTrue(self.first_child.rgt==3)
|
|
self.assertTrue(self.first_sibling.lft==4)
|
|
self.assertTrue(self.first_sibling.rgt==5)
|
|
self.assertTrue(self.root.rgt==6)
|
|
|
|
def test_remove_sibling(self):
|
|
self.test_sibling()
|
|
self.first_sibling.parent_node = ''
|
|
update_nsm(self.first_sibling)
|
|
|
|
self.root._loadfromdb()
|
|
self.first_child._loadfromdb()
|
|
|
|
self.assertTrue(self.root.lft==1)
|
|
self.assertTrue(self.first_child.lft==2)
|
|
self.assertTrue(self.first_child.rgt==3)
|
|
self.assertTrue(self.root.rgt==4)
|
|
self.assertTrue(self.first_sibling.lft==5)
|
|
self.assertTrue(self.first_sibling.rgt==6)
|
|
|
|
def test_change_parent(self):
|
|
self.test_sibling()
|
|
|
|
# add grand child
|
|
self.grand_child.save(1)
|
|
update_nsm(self.grand_child)
|
|
|
|
# check lft rgt
|
|
self.assertTrue(self.grand_child.lft==3)
|
|
self.assertTrue(self.grand_child.rgt==4)
|
|
|
|
# change parent
|
|
self.grand_child.parent_node = 'C002'
|
|
self.grand_child.save()
|
|
|
|
# update
|
|
update_nsm(self.grand_child)
|
|
|
|
# check lft rgt
|
|
self.assertTrue(self.grand_child.lft==5)
|
|
self.assertTrue(self.grand_child.rgt==6)
|
|
|
|
|
|
def tearDown(self):
|
|
webnotes.conn.sql("drop table tabnsmtest")
|
|
|
|
|
|
|
|
# called in the on_update method
|
|
def update_nsm(doc_obj):
|
|
# get fields, data from the DocType
|
|
|
|
pf, opf = 'parent_node', 'old_parent'
|
|
|
|
if str(doc_obj.__class__)=='webnotes.model.doc.Document':
|
|
# passed as a Document object
|
|
d = doc_obj
|
|
else:
|
|
# passed as a DocType object
|
|
d = doc_obj.doc
|
|
if hasattr(doc_obj,'nsm_parent_field'):
|
|
pf = doc_obj.nsm_parent_field
|
|
if hasattr(doc_obj,'nsm_oldparent_field'):
|
|
opf = doc_obj.nsm_oldparent_field
|
|
|
|
p, op = d.fields.get(pf, ''), d.fields.get(opf, '')
|
|
|
|
# has parent changed (?) or parent is None (root)
|
|
if not d.lft and not d.rgt:
|
|
update_add_node(d.doctype, d.name, p or '', pf)
|
|
elif op != p:
|
|
update_remove_node(d.doctype, d.name)
|
|
update_add_node(d.doctype, d.name, p or '', pf)
|
|
|
|
# set old parent
|
|
webnotes.conn.set(d, opf, p or '')
|
|
|
|
# reload
|
|
d._loadfromdb()
|
|
|
|
def rebuild_tree(doctype, parent_field):
|
|
"""
|
|
call rebuild_node for all root nodes
|
|
"""
|
|
# get all roots
|
|
right = 1
|
|
result = webnotes.conn.sql("SELECT name FROM `tab%s` WHERE `%s`='' or `%s` IS NULL ORDER BY name ASC" % (doctype, parent_field, parent_field))
|
|
for r in result:
|
|
right = rebuild_node(doctype, r[0], right, parent_field)
|
|
webnotes.conn.sql("commit")
|
|
webnotes.conn.sql("start transaction")
|
|
|
|
def rebuild_node(doctype, parent, left, parent_field, cnt = 0):
|
|
"""
|
|
reset lft, rgt and recursive call for all children
|
|
"""
|
|
from webnotes.utils import now
|
|
n = now()
|
|
|
|
# the right value of this node is the left value + 1
|
|
right = left+1
|
|
|
|
# get all children of this node
|
|
result = webnotes.conn.sql("SELECT name FROM `tab%s` WHERE `%s`='%s'" % (doctype, parent_field, parent))
|
|
for r in result:
|
|
right = rebuild_node(doctype, r[0], right, parent_field, cnt)
|
|
|
|
# we've got the left value, and now that we've processed
|
|
# the children of this node we also know the right value
|
|
webnotes.conn.sql("UPDATE `tab%s` SET lft=%s, rgt=%s, modified='%s' WHERE name='%s'" % (doctype,left,right,n,parent))
|
|
|
|
# commit after every 100
|
|
cnt += 1
|
|
if cnt % 100 == 0:
|
|
cnt = 0
|
|
webnotes.conn.sql("commit")
|
|
webnotes.conn.sql("start transaction")
|
|
|
|
#return the right value of this node + 1
|
|
return right+1
|
|
|
|
def update_add_node(doctype, name, parent, parent_field):
|
|
"""
|
|
insert a new node
|
|
"""
|
|
from webnotes.utils import now
|
|
n = now()
|
|
|
|
# get the last sibling of the parent
|
|
if parent:
|
|
right = webnotes.conn.sql("select rgt from `tab%s` where name='%s'" % (doctype, parent))[0][0]
|
|
else: # root
|
|
right = webnotes.conn.sql("select ifnull(max(rgt),0)+1 from `tab%s` where ifnull(`%s`,'') =''" % (doctype, parent_field))[0][0]
|
|
right = right or 1
|
|
|
|
# update all on the right
|
|
webnotes.conn.sql("update `tab%s` set rgt = rgt+2, modified='%s' where rgt >= %s" %(doctype,n,right))
|
|
webnotes.conn.sql("update `tab%s` set lft = lft+2, modified='%s' where lft >= %s" %(doctype,n,right))
|
|
|
|
# update index of new node
|
|
if webnotes.conn.sql("select * from `tab%s` where lft=%s or rgt=%s"% (doctype, right, right+1)):
|
|
webnotes.msgprint("Nested set error. Please send mail to support")
|
|
raise Exception
|
|
|
|
webnotes.conn.sql("update `tab%s` set lft=%s, rgt=%s, modified='%s' where name='%s'" % (doctype,right,right+1,n,name))
|
|
return right
|
|
|
|
def update_remove_node(doctype, name):
|
|
"""
|
|
remove a node
|
|
"""
|
|
from webnotes.utils import now
|
|
n = now()
|
|
|
|
left = webnotes.conn.sql("select lft from `tab%s` where name='%s'" % (doctype,name))
|
|
if left[0][0]:
|
|
# reset this node
|
|
webnotes.conn.sql("update `tab%s` set lft=0, rgt=0, modified='%s' where name='%s'" % (doctype,n,name))
|
|
|
|
# update all on the right
|
|
webnotes.conn.sql("update `tab%s` set rgt = rgt-2, modified='%s' where rgt > %s" %(doctype,n,left[0][0]))
|
|
webnotes.conn.sql("update `tab%s` set lft = lft-2, modified='%s' where lft > %s" %(doctype,n,left[0][0]))
|