[Zope-CVS] CVS: Packages/pypes/pypes/tests - test_query.py:1.2

[Casey Duncan]

Update of /cvs-repository/Packages/pypes/pypes/tests
In directory cvs.zope.org:/tmp/cvs-serv24863/tests

Modified Files:
	test_query.py 
Log Message:
Implement equi-join query primitive


=== Packages/pypes/pypes/tests/test_query.py 1.1 => 1.2 ===
--- Packages/pypes/pypes/tests/test_query.py:1.1	Tue Mar 30 23:58:09 2004
+++ Packages/pypes/pypes/tests/test_query.py	Fri Apr  2 22:50:24 2004
@@ -144,7 +144,91 @@
             sort(self.scrambled, lambda x: x.foo, descending, limit=3))
         self.ordered.reverse()
         self.assertEqual(sorted, self.ordered[:3])
+        
+    def test_sort_empty(self):
+        from pypes.query import sort, descending
+        sorted = sort([], lambda x:x)
+        self.failIf(sorted)
+        sorted = sort([], lambda x:x, descending)
+        self.failIf(sorted)
+        sorted = sort([], lambda x:x, limit=20)
+        self.failIf(sorted)
+        sorted = sort([], lambda x:x, descending, limit=5)
+        self.failIf(sorted)
+
+class TestJoin(unittest.TestCase):
+    
+    def test_equijoin_one2one(self):
+        from pypes.query import equijoin
+        us = [TestClass(age=i+10) for i in range(15)]
+        them = [TestClass(age=14-i) for i in range(10)]
+        expr = lambda x: x.age
+        joined = list(equijoin(us, expr, them, expr))
+        self.assertEqual(len(joined), 5)
+        for left, right in joined:
+            self.assertEqual(left.age, right.age)
+            
+    def test_equijoin_empty(self):
+        from pypes.query import equijoin
+        joined = equijoin([], lambda x:x, [], lambda x:x)
+        self.failIf(list(joined))
+        objs = [TestClass() for i in range(5)]
+        joined = equijoin(objs, lambda x:x, [], lambda x:x)
+        self.failIf(list(joined))
+        joined = equijoin([], lambda x:x, objs, lambda x:x)
+        self.failIf(list(joined))
+    
+    def test_equijoin_none_match(self):
+        from pypes.query import equijoin
+        objs = [TestClass() for i in range(5)]
+        joined = equijoin(objs, lambda x: 1, objs, lambda x: -1)
+        self.failIf(list(joined))                
             
+    def test_equijoin_one2many(self):
+        from sets import Set
+        from pypes.query import equijoin
+        colors = ['red', 'orange', 'yellow', 'green']
+        left = Set([TestClass(shade=c) for c in colors])
+        right = Set()
+        for i in range(4):
+            for c in colors[1:]:
+                right.union_update([TestClass(shade=c) for k in range(i)])
+        shade = lambda x: x.shade
+        joined_right = Set()
+        joined_left = Set()
+        for l, r in equijoin(left, shade, right, shade):
+            self.assertEqual(l.shade, r.shade)
+            self.assertNotEqual(l.shade, 'red')
+            joined_left.add(l)
+            joined_right.add(r)
+        self.assertEqual(len(joined_left), len(left) - 1)
+        self.assertEqual(joined_right, right)
+        # Test left <=> right switch optimization
+        for r, l in equijoin(right, shade, left, shade):
+            self.assertEqual(l.shade, r.shade)
+            self.assertNotEqual(l.shade, 'red')
+            joined_left.add(l)
+            joined_right.add(r)
+        self.assertEqual(len(joined_left), len(left) - 1)
+        self.assertEqual(joined_right, right)
     
+    def test_equijoin_many2many(self):
+        from pypes.query import equijoin
+        big = [TestClass(big=True) for i in range(10)]
+        small = [TestClass(big=False) for i in range(10)]
+        left = big[4:] + small[4:]
+        right = big[:4] + small[:4]
+        isbig = lambda x: x.big
+        join = list(equijoin(left, isbig, right, isbig))
+        self.assertEqual(len(join), len(left) * len(right) // 2)
+        for r, l in join:
+            self.assertEqual(l.big, r.big)
+        # Test left <=> right switch optimization
+        join = list(equijoin(right, isbig, left, isbig))
+        self.assertEqual(len(join), len(left) * len(right) // 2)
+        for r, l in join:
+            self.assertEqual(l.big, r.big)
+
+
 if __name__ == '__main__':
     unittest.main()