- added sqlalchemy.ext.orderinglist, a custom list class that synchronizes an object attribute with that object's position in the list

1 files changed, 422 insertions, 0 deletions

diff --git a/test/ext/orderinglist.py b/test/ext/orderinglist.py
new file mode 100644
index 000000000..73d0405a4
--- /dev/null
+++ b/test/ext/orderinglist.py
@@ -0,0 +1,422 @@
+from testbase import PersistTest
+import sqlalchemy.util as util
+import unittest, sys, os
+import testbase
+from sqlalchemy import *
+from sqlalchemy.ext.orderinglist import *
+
+db = testbase.db
+metadata = None
+
+# order in whole steps 
+def step_numbering(step):
+    def f(index, collection):
+        return step * index
+    return f
+
+# almost fibonacci- skip the first 2 steps
+# e.g. 1, 2, 3, 5, 8, ... instead of 0, 1, 1, 2, 3, ...
+# otherwise ordering of the elements at '1' is undefined... ;)
+def fibonacci_numbering(order_col):
+    def f(index, collection):
+        if index == 0:
+            return 1
+        elif index == 1:
+            return 2
+        else:
+            return (getattr(collection[index - 1], order_col) +
+                    getattr(collection[index - 2], order_col))
+    return f
+
+# 0 -> A, 1 -> B, ... 25 -> Z, 26 -> AA, 27 -> AB, ...
+def alpha_ordering(index, collection):
+    s = ''
+    while index > 25:
+        d = index / 26
+        s += chr((d % 26) + 64)
+        index -= d * 26
+    s += chr(index + 65)
+    return s
+
+class OrderingListTest(PersistTest):
+    def setUp(self):
+        global metadata, slides_table, bullets_table, Slide, Bullet
+        slides_table, bullets_table = None, None
+        Slide, Bullet = None, None
+        if metadata:
+            metadata.clear()
+
+    def _setup(self, test_collection_class):
+        """Build a relation situation using the given test_collection_class
+        factory"""
+
+        global metadata, slides_table, bullets_table, Slide, Bullet
+
+        metadata = BoundMetaData(db)
+        slides_table = Table('test_Slides', metadata,
+                             Column('id', Integer, primary_key=True),
+                             Column('name', String))
+        bullets_table = Table('test_Bullets', metadata,
+                              Column('id', Integer, primary_key=True),
+                              Column('slide_id', Integer,
+                                     ForeignKey('test_Slides.id')),
+                              Column('position', Integer),
+                              Column('text', String))
+
+        class Slide(object):
+            def __init__(self, name):
+                self.name = name
+            def __repr__(self):
+                return '<Slide "%s">' % self.name
+
+        class Bullet(object):
+            def __init__(self, text):
+                self.text = text
+            def __repr__(self):
+                return '<Bullet "%s" pos %s>' % (self.text, self.position)
+
+        mapper(Slide, slides_table, properties={
+            'bullets': relation(Bullet, lazy=False,
+                                collection_class=test_collection_class,
+                                backref='slide',
+                                order_by=[bullets_table.c.position])
+            })
+        mapper(Bullet, bullets_table)
+
+        metadata.create_all()
+
+    def tearDown(self):
+        metadata.drop_all()
+
+    def test_append_no_reorder(self):
+        self._setup(ordering_list('position', count_from=1,
+                                  reorder_on_append=False))
+
+        s1 = Slide('Slide #1')
+
+        self.assert_(not s1.bullets)
+        self.assert_(len(s1.bullets) == 0)
+
+        s1.bullets.append(Bullet('s1/b1'))
+
+        self.assert_(s1.bullets)
+        self.assert_(len(s1.bullets) == 1)
+        self.assert_(s1.bullets[0].position == 1)
+
+        s1.bullets.append(Bullet('s1/b2'))
+
+        self.assert_(len(s1.bullets) == 2)
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+
+        bul = Bullet('s1/b100')
+        bul.position = 100
+        s1.bullets.append(bul)
+
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+        self.assert_(s1.bullets[2].position == 100)
+
+        s1.bullets.append(Bullet('s1/b4'))
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+        self.assert_(s1.bullets[2].position == 100)
+        self.assert_(s1.bullets[3].position == 4)
+
+        s1.bullets._reorder()
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+        self.assert_(s1.bullets[2].position == 3)
+        self.assert_(s1.bullets[3].position == 4)
+
+        session = create_session()
+        session.save(s1)
+        session.flush()
+
+        id = s1.id
+        session.clear()
+        del s1
+
+        srt = session.query(Slide).get(id)
+
+        self.assert_(srt.bullets)
+        self.assert_(len(srt.bullets) == 4)
+    
+        titles = ['s1/b1','s1/b2','s1/b100','s1/b4']
+        found = [b.text for b in srt.bullets]
+
+        self.assert_(titles == found)
+
+    def test_append_reorder(self):
+        self._setup(ordering_list('position', count_from=1,
+                                  reorder_on_append=True))
+
+        s1 = Slide('Slide #1')
+
+        self.assert_(not s1.bullets)
+        self.assert_(len(s1.bullets) == 0)
+
+        s1.bullets.append(Bullet('s1/b1'))
+
+        self.assert_(s1.bullets)
+        self.assert_(len(s1.bullets) == 1)
+        self.assert_(s1.bullets[0].position == 1)
+
+        s1.bullets.append(Bullet('s1/b2'))
+
+        self.assert_(len(s1.bullets) == 2)
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+
+        bul = Bullet('s1/b100')
+        bul.position = 100
+        s1.bullets.append(bul)
+
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+        self.assert_(s1.bullets[2].position == 3)
+            
+        s1.bullets.append(Bullet('s1/b4'))
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+        self.assert_(s1.bullets[2].position == 3)
+        self.assert_(s1.bullets[3].position == 4)
+
+        s1.bullets._reorder()
+        self.assert_(s1.bullets[0].position == 1)
+        self.assert_(s1.bullets[1].position == 2)
+        self.assert_(s1.bullets[2].position == 3)
+        self.assert_(s1.bullets[3].position == 4)
+
+        session = create_session()
+        session.save(s1)
+        session.flush()
+
+        id = s1.id
+        session.clear()
+        del s1
+
+        srt = session.query(Slide).get(id)
+
+        self.assert_(srt.bullets)
+        self.assert_(len(srt.bullets) == 4)
+
+        titles = ['s1/b1','s1/b2','s1/b100','s1/b4']
+        found = [b.text for b in srt.bullets]
+
+        self.assert_(titles == found)
+                
+    def test_insert(self):
+        self._setup(ordering_list('position'))
+
+        s1 = Slide('Slide #1')
+        s1.bullets.append(Bullet('1'))
+        s1.bullets.append(Bullet('2'))
+        s1.bullets.append(Bullet('3'))
+        s1.bullets.append(Bullet('4'))
+
+        self.assert_(s1.bullets[0].position == 0)
+        self.assert_(s1.bullets[1].position == 1)
+        self.assert_(s1.bullets[2].position == 2)
+        self.assert_(s1.bullets[3].position == 3)
+    
+        s1.bullets.insert(2, Bullet('insert_at_2'))
+        self.assert_(s1.bullets[0].position == 0)
+        self.assert_(s1.bullets[1].position == 1)
+        self.assert_(s1.bullets[2].position == 2)
+        self.assert_(s1.bullets[3].position == 3)
+        self.assert_(s1.bullets[4].position == 4)
+
+        self.assert_(s1.bullets[1].text == '2')
+        self.assert_(s1.bullets[2].text == 'insert_at_2')
+        self.assert_(s1.bullets[3].text == '3')
+
+        s1.bullets.insert(999, Bullet('999'))
+
+        self.assert_(len(s1.bullets) == 6)
+        self.assert_(s1.bullets[5].position == 5)
+
+        session = create_session()
+        session.save(s1)
+        session.flush()
+
+        id = s1.id
+        session.clear()
+        del s1
+
+        srt = session.query(Slide).get(id)
+
+        self.assert_(srt.bullets)
+        self.assert_(len(srt.bullets) == 6)
+    
+        texts = ['1','2','insert_at_2','3','4','999']
+        found = [b.text for b in srt.bullets]
+
+        self.assert_(texts == found)
+
+    def test_slice(self):
+        self._setup(ordering_list('position'))
+
+        b = [ Bullet('1'), Bullet('2'), Bullet('3'),
+              Bullet('4'), Bullet('5'), Bullet('6') ]
+        s1 = Slide('Slide #1')
+
+        # 1, 2, 3
+        s1.bullets[0:3] = b[0:3]
+        for i in 0, 1, 2:
+            self.assert_(s1.bullets[i].position == i)
+            self.assert_(s1.bullets[i] == b[i])
+
+        # 1, 4, 5, 6, 3
+        s1.bullets[1:2] = b[3:6]
+        for li, bi in (0,0), (1,3), (2,4), (3,5), (4,2):
+            self.assert_(s1.bullets[li].position == li)
+            self.assert_(s1.bullets[li] == b[bi])
+
+        # 1, 6, 3
+        del s1.bullets[1:3]
+        for li, bi in (0,0), (1,5), (2,2):
+            self.assert_(s1.bullets[li].position == li)
+            self.assert_(s1.bullets[li] == b[bi])
+
+        session = create_session()
+        session.save(s1)
+        session.flush()
+
+        id = s1.id
+        session.clear()
+        del s1
+
+        srt = session.query(Slide).get(id)
+
+        self.assert_(srt.bullets)
+        self.assert_(len(srt.bullets) == 3)
+    
+        texts = ['1', '6', '3']
+        for i, text in enumerate(texts):
+            self.assert_(srt.bullets[i].position == i)
+            self.assert_(srt.bullets[i].text == text)
+
+    def test_replace1(self):
+        self._setup(ordering_list('position'))
+        
+        s1 = Slide('Slide #1')
+        s1.bullets = [ Bullet('1'), Bullet('2'), Bullet('3') ]
+
+        self.assert_(len(s1.bullets) == 3)
+        self.assert_(s1.bullets[2].position == 2)
+
+        session = create_session()
+        session.save(s1)
+        session.flush()
+
+        new_bullet = Bullet('new 2')
+        self.assert_(new_bullet.position is None)
+
+        # naive replacement, no database deletion should occur
+        # with current InstrumentedList __setitem__ semantics
+        s1.bullets[1] = new_bullet
+
+        self.assert_(new_bullet.position == 1)
+        self.assert_(len(s1.bullets) == 3)
+
+        id = s1.id
+
+        session.flush()
+        session.clear()
+
+        srt = session.query(Slide).get(id)
+    
+        self.assert_(srt.bullets)
+        self.assert_(len(srt.bullets) == 4)
+
+        self.assert_(srt.bullets[1].text == '2')
+        self.assert_(srt.bullets[2].text == 'new 2')
+    
+    def test_replace2(self):
+        self._setup(ordering_list('position'))
+
+        s1 = Slide('Slide #1')
+        s1.bullets = [ Bullet('1'), Bullet('2'), Bullet('3') ]
+
+        self.assert_(len(s1.bullets) == 3)
+        self.assert_(s1.bullets[2].position == 2)
+
+        session = create_session()
+        session.save(s1)
+        session.flush()
+
+        new_bullet = Bullet('new 2')
+        self.assert_(new_bullet.position is None)
+
+        # mark existing bullet as db-deleted before replacement.
+        session.delete(s1.bullets[1])
+        s1.bullets[1] = new_bullet
+
+        self.assert_(new_bullet.position == 1)
+        self.assert_(len(s1.bullets) == 3)
+
+        id = s1.id
+
+        session.flush()
+        session.clear()
+
+        srt = session.query(Slide).get(id)
+    
+        self.assert_(srt.bullets)
+        self.assert_(len(srt.bullets) == 3)
+
+        self.assert_(srt.bullets[1].text == 'new 2')
+        self.assert_(srt.bullets[2].text == '3')
+    
+    def test_funky_ordering(self):
+        class Pos(object):
+            def __init__(self):
+                self.position = None
+
+        step_factory = ordering_list('position',
+                                     ordering_func=step_numbering(2))
+
+        stepped = step_factory()
+        stepped.append(Pos())
+        stepped.append(Pos())
+        stepped.append(Pos())
+        stepped.append(Pos())
+
+        for li, pos in (0,0), (1,2), (2,4), (3,6):
+            self.assert_(stepped[li].position == pos)
+
+        fib_factory = ordering_list('position',
+          ordering_func=fibonacci_numbering('position'))
+
+        fibbed = fib_factory()
+        fibbed.append(Pos())
+        fibbed.append(Pos())
+        fibbed.append(Pos())
+        fibbed.append(Pos())
+        fibbed.append(Pos())
+
+        for li, pos in (0,1), (1,2), (2,3), (3,5), (4,8):
+            self.assert_(fibbed[li].position == pos)
+
+        fibbed.insert(2, Pos())
+        fibbed.insert(4, Pos())
+        fibbed.insert(6, Pos())
+    
+        for li, pos in (0,1), (1,2), (2,3), (3,5), (4,8), (5,13), (6,21), (7,34):
+            self.assert_(fibbed[li].position == pos)
+
+        alpha_factory = ordering_list('position',
+                                      ordering_func=alpha_ordering)
+        alpha = alpha_factory()
+        alpha.append(Pos())
+        alpha.append(Pos())
+        alpha.append(Pos())
+
+        alpha.insert(1, Pos())
+
+        for li, pos in (0,'A'), (1,'B'), (2,'C'), (3,'D'):
+            self.assert_(alpha[li].position == pos)
+
+if __name__ == "__main__":
+    testbase.main()