Forward port patch from collector #1118 (sqlgroup support for UPDATE).

lib/python/Shared/DC/ZRDB/sqlgroup.py

@@ -11,7 +11,7 @@
 #
 ##############################################################################
 
-'''Inserting optional tests with 'sqlgroup'
+"""Inserting optional tests with 'sqlgroup'
 
     It is sometimes useful to make inputs to an SQL statement
     optinal.  Doing so can be difficult, because not only must the
@@ -59,7 +59,9 @@
     other than whitespace characters.  If it does, then it is inserted
     with the appropriate boolean operator, as indicated by use of an
     'and' or 'or' tag, otherwise, no text is inserted.
-'''
+
+$Id$
+"""
 
 ############################################################################
 #     Copyright
@@ -73,40 +75,62 @@ __rcs_id__='$Id$'
 __version__='$Revision: 1.10 $'[11:-2]
 
 from DocumentTemplate.DT_Util import parse_params
-str=__builtins__['str']
+str = __builtins__['str']
 from string import strip, join
 import sys
 
+_TNAME_MAPPING = {'comma': ','}
+
 class SQLGroup:
-    blockContinuations='and','or'
-    name='sqlgroup'
-    required=None
-    where=None
+    blockContinuations = 'and', 'or', 'comma'
+    name = 'sqlgroup'
+    required = None
+    where = None
+    set = None
+    noparens = None
 
     def __init__(self, blocks):
 
-        self.blocks=blocks
+        self.blocks = blocks
         tname, args, section = blocks[0]
-        self.__name__="%s %s" % (tname, args)
-        args = parse_params(args, required=1, where=1)
-        if args.has_key(''): args[args['']]=1
-        if args.has_key('required'): self.required=args['required']
-        if args.has_key('where'): self.where=args['where']
+        self.__name__ = "%s %s" % (tname, args)
+        args = parse_params(args, required=1, where=1, set=1, noparens=1)
+        if args.has_key(''):
+            args[args['']] = 1
+        if args.has_key('required'):
+            self.required = args['required']
+        if args.has_key('where'):
+            self.where = args['where']
+        if args.has_key('set'):
+            self.set = args['set']
+        if args.has_key('noparens'):
+            self.noparens = args['noparens']
 
     def render(self,md):
 
-        r=[]
+        r = []
         for tname, args, section in self.blocks:
-            __traceback_info__=tname
-            s=strip(section(None, md))
+            __traceback_info__ = tname
+            s = strip(section(None, md))
             if s:
-                if r: r.append(tname)
-                r.append("%s\n" % s)
-
+                if r:
+                    r.append(_TNAME_MAPPING.get(tname, tname))
+                if self.noparens:
+                    r.append(s)
+                else:
+                    r.append("%s\n" % s)
         if r:
-            if len(r) > 1: r="(%s)\n" % join(r,' ')
-            else: r=r[0]
-            if self.where: r="where\n"+r
+            if len(r) > 1:
+                if self.noparens:
+                    r = "%s\n" % join(r,' ')
+                else:
+                    r = "(%s)\n" % join(r,' ')
+            else:
+                r = r[0]
+            if self.set:
+                r = "set\n"+r
+            if self.where:
+                r = "where\n"+r
             return r
 
         if self.required:
@@ -114,4 +138,4 @@ class SQLGroup:
 
         return ''
 
-    __call__=render
+    __call__ = render

lib/python/Shared/DC/ZRDB/tests/test_sqlgroup.py

+##############################################################################
+#
+# Copyright (c) 2005 Zope Corporation and Contributors. All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+import unittest
+
+from UserDict import UserDict
+
+def _sql_quote(v):
+    return '"%s"' % v
+
+class SQLGroupTests(unittest.TestCase):
+
+    def _getTargetClass(self):
+        from Shared.DC.ZRDB.sqlgroup import SQLGroup
+        return SQLGroup
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_empty_args(self):
+        group = self._makeOne([('sqlgroup', '', None)])
+        self.assertEqual(group.__name__, 'sqlgroup ')
+        self.failIf(group.required)
+        self.failIf(group.where)
+        self.failIf(group.set)
+        self.failIf(group.noparens)
+
+    def test_ctor_required(self):
+        group = self._makeOne([('sqlgroup', 'required', None)])
+        self.assertEqual(group.__name__, 'sqlgroup required')
+        self.failUnless(group.required)
+        self.failIf(group.where)
+        self.failIf(group.set)
+        self.failIf(group.noparens)
+
+    def test_ctor_where(self):
+        group = self._makeOne([('sqlgroup', 'where', None)])
+        self.assertEqual(group.__name__, 'sqlgroup where')
+        self.failIf(group.required)
+        self.failUnless(group.where)
+        self.failIf(group.set)
+        self.failIf(group.noparens)
+
+    def test_ctor_noparens(self):
+        group = self._makeOne([('sqlgroup', 'noparens', None)])
+        self.assertEqual(group.__name__, 'sqlgroup noparens')
+        self.failIf(group.required)
+        self.failIf(group.where)
+        self.failIf(group.set)
+        self.failUnless(group.noparens)
+
+    def test_ctor_set(self):
+        group = self._makeOne([('sqlgroup', 'set', None)])
+        self.assertEqual(group.__name__, 'sqlgroup set')
+        self.failIf(group.required)
+        self.failIf(group.where)
+        self.failUnless(group.set)
+        self.failIf(group.noparens)
+
+    def test_render_empty_optional(self):
+        group = self._makeOne([('sqlgroup', '', lambda x, y:'')])
+        md = {}
+        self.assertEqual(group.render(md), '')
+
+    def test_render_empty_optional_where(self):
+        group = self._makeOne([('sqlgroup', 'where', lambda x, y:'')])
+        md = {}
+        self.assertEqual(group.render(md), '')
+
+    def test_render_empty_optional_set(self):
+        group = self._makeOne([('sqlgroup', 'set', lambda x, y:'')])
+        md = {}
+        self.assertEqual(group.render(md), '')
+
+    def test_render_empty_required_raises_ValueError(self):
+        group = self._makeOne([('sqlgroup', 'required', lambda x, y:'')])
+        md = {}
+        self.assertRaises(ValueError, group.render, md)
+
+    def test_render_one_block(self):
+        group = self._makeOne([('sqlgroup', '', lambda x, y:'abc'),
+                              ])
+        md = {}
+        rendered = group.render(md)
+        rendered = ''.join(rendered.split('\n'))
+        self.assertEqual(rendered, 'abc')
+
+    def test_render_one_block_where(self):
+        group = self._makeOne([('sqlgroup', 'where', lambda x, y:'abc'),
+                              ])
+        md = {}
+        rendered = group.render(md)
+        self.assertEqual(rendered, 'where\nabc\n')
+
+    def test_render_one_block_set(self):
+        group = self._makeOne([('sqlgroup', 'set', lambda x, y:'abc'),
+                              ])
+        md = {}
+        rendered = group.render(md)
+        self.assertEqual(rendered, 'set\nabc\n')
+
+    def test_render_multiple_blocks_with_tname(self):
+        group = self._makeOne([('sqlgroup', '', lambda x, y:'abc'),
+                               ('baz', '', lambda x, y: 'def'),
+                               ('qux', '', lambda x, y: 'ghi'),
+                              ])
+        md = {}
+        rendered = group.render(md)
+        rendered = ''.join(rendered.split('\n'))
+        self.assertEqual(rendered, '(abc baz def qux ghi)')
+
+    def test_render_multiple_blocks_with_tname_noparens(self):
+        group = self._makeOne([('sqlgroup', 'noparens', lambda x, y:'abc'),
+                               ('baz', '', lambda x, y: 'def'),
+                               ('qux', '', lambda x, y: 'ghi'),
+                              ])
+        md = {}
+        rendered = group.render(md)
+        rendered = ''.join(rendered.split('\n'))
+        self.assertEqual(rendered, 'abc baz def qux ghi')
+
+    def test_render_multiple_blocks_with_tname_and_where(self):
+        group = self._makeOne([('sqlgroup', 'where', lambda x, y:'abc'),
+                               ('baz', '', lambda x, y: 'def'),
+                               ('qux', '', lambda x, y: 'ghi'),
+                              ])
+        md = {}
+        rendered = group.render(md)
+        rendered = ''.join(rendered.split('\n'))
+        self.assertEqual(rendered, 'where(abc baz def qux ghi)')
+
+
+    def test_parsed_rendered_complex_where(self):
+        # something of a functional test, as we use nvSQL to get parsed.
+        from Shared.DC.ZRDB.DA import nvSQL
+        template = nvSQL(WHERE_EXAMPLE)
+        mapping = {}
+        mapping['name'] = 'Goofy'
+        mapping['home_town'] = 'Orlando'
+        mapping['sql_quote__'] = _sql_quote
+
+        rendered = template(None, mapping)
+        self.assertEqual(rendered,
+                         'select * from actors\n'
+                         'where\n'
+                         '((nick_name = "Goofy"\n'
+                         ' or first_name = "Goofy"\n)\n'
+                         ' and home_town = "Orlando"\n)\n'
+                        )
+
+    def test_parsed_rendered_complex_set(self):
+        # something of a functional test, as we use nvSQL to get parsed.
+        from Shared.DC.ZRDB.DA import nvSQL
+        template = nvSQL(UPDATE_EXAMPLE)
+        mapping = {}
+        mapping['nick_name'] = 'Goofy'
+        mapping['home_town'] = 'Orlando'
+        mapping['sql_quote__'] = _sql_quote
+
+        rendered = template(None, mapping)
+        self.assertEqual(rendered,
+                         'update actors\n'
+                         'set\nnick_name = "Goofy" , home_town = "Orlando"\n'
+                        )
+
+WHERE_EXAMPLE = """\
+select * from actors
+<dtml-sqlgroup where required>
+  <dtml-sqlgroup>
+    <dtml-sqltest name column=nick_name type=nb multiple optional>
+  <dtml-or>
+    <dtml-sqltest name column=first_name type=nb multiple optional>
+  </dtml-sqlgroup>
+<dtml-and>
+  <dtml-sqltest home_town type=nb optional>
+<dtml-and>
+  <dtml-if minimum_age>
+     age >= <dtml-sqlvar minimum_age type=int>
+  </dtml-if>
+<dtml-and>
+  <dtml-if maximum_age>
+     age <= <dtml-sqlvar maximum_age type=int>
+  </dtml-if>
+</dtml-sqlgroup>
+"""
+
+UPDATE_EXAMPLE = """\
+update actors
+<dtml-sqlgroup set noparens>
+<dtml-sqltest nick_name type=nb optional>
+<dtml-comma>
+<dtml-sqltest home_town type=nb optional>
+</dtml-sqlgroup>
+"""
+
+def test_suite():
+    suite = unittest.TestSuite()
+    suite.addTest(unittest.makeSuite(SQLGroupTests))
+    return suite
+
+if __name__ == '__main__':
+    unittest.main(defaultTest='test_suite')