Merge pull request #470 from hbrunn/8.0-database_cleanup_backport_9_improvements

8.0 database cleanup backport 9 improvements

database_cleanup/identifier_adapter.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+# © 2016 Therp BV <http://therp.nl>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
+from psycopg2.extensions import ISQLQuote
+
+
+class IdentifierAdapter(ISQLQuote):
+    def __init__(self, identifier, quote=True):
+        self.quote = quote
+        self.identifier = identifier
+
+    def __conform__(self, protocol):
+        if protocol == ISQLQuote:
+            return self
+
+    def getquoted(self):
+        def is_identifier_char(c):
+            return c.isalnum() or c in ['_', '$']
+
+        format_string = '"%s"'
+        if not self.quote:
+            format_string = '%s'
+        return format_string % filter(is_identifier_char, self.identifier)

database_cleanup/model/purge_columns.py

@@ -21,6 +21,7 @@
 
 from openerp.osv import orm, fields
 from openerp.tools.translate import _
+from ..identifier_adapter import IdentifierAdapter
 
 
 class CleanupPurgeLineColumn(orm.TransientModel):
@@ -61,9 +62,11 @@ class CleanupPurgeLineColumn(orm.TransientModel):
                 'Dropping column %s from table %s',
                 line.name, model_pool._table)
             cr.execute(
-                """
-                ALTER TABLE "%s" DROP COLUMN "%s"
-                """ % (model_pool._table, line.name))
+                "ALTER TABLE %s DROP COLUMN %s",
+                (
+                    IdentifierAdapter(model_pool._table),
+                    IdentifierAdapter(line.name),
+                ))
             line.write({'purged': True})
             cr.commit()
         return True

database_cleanup/model/purge_data.py

@@ -21,6 +21,7 @@
 
 from openerp.osv import orm, fields
 from openerp.tools.translate import _
+from ..identifier_adapter import IdentifierAdapter
 
 
 class CleanupPurgeLineData(orm.TransientModel):
@@ -80,11 +81,11 @@ class CleanupPurgeWizardData(orm.TransientModel):
             cr.execute(
                 """
                 SELECT id FROM ir_model_data
-                WHERE model = %%s
+                WHERE model = %s
                 AND res_id IS NOT NULL
                 AND NOT EXISTS (
                     SELECT id FROM %s WHERE id=ir_model_data.res_id)
-                """ % self.pool[model]._table, (model,))
+                """, (model, IdentifierAdapter(self.pool[model]._table)))
             data_ids += [data_row[0] for data_row in cr.fetchall()]
         data_ids += data_pool.search(
             cr, uid, [('model', 'in', unknown_models)], context=context)

database_cleanup/model/purge_tables.py

@@ -21,6 +21,7 @@
 
 from openerp.osv import orm, fields
 from openerp.tools.translate import _
+from ..identifier_adapter import IdentifierAdapter
 
 
 class CleanupPurgeLineTable(orm.TransientModel):
@@ -62,7 +63,7 @@ class CleanupPurgeLineTable(orm.TransientModel):
                 WHERE af.attnum = confkey AND af.attrelid = confrelid AND
                 a.attnum = conkey AND a.attrelid = conrelid
                 AND confrelid::regclass = '%s'::regclass;
-                """ % line.name)
+                """, (IdentifierAdapter(line.name, quote=False),))
 
             for constraint in cr.fetchall():
                 if constraint[3] in tables:
@@ -70,12 +71,15 @@ class CleanupPurgeLineTable(orm.TransientModel):
                         'Dropping constraint %s on table %s (to be dropped)',
                         constraint[0], constraint[3])
                     cr.execute(
-                        "ALTER TABLE %s DROP CONSTRAINT %s" % (
-                            constraint[3], constraint[0]))
+                        "ALTER TABLE %s DROP CONSTRAINT %s", (
+                            IdentifierAdapter(constraint[3]),
+                            IdentifierAdapter(constraint[0]),
+                        )
+                    )
 
             self.logger.info(
                 'Dropping table %s', line.name)
-            cr.execute("DROP TABLE \"%s\"" % (line.name,))
+            cr.execute("DROP TABLE %s", (IdentifierAdapter(line.name),))
             line.write({'purged': True})
             cr.commit()
         return True
@@ -116,13 +120,11 @@ class CleanupPurgeWizardTable(orm.TransientModel):
                 ]
 
         # Cannot pass table names as a psycopg argument
-        known_tables_repr = ",".join(
-            [("'%s'" % table) for table in known_tables])
         cr.execute(
             """
             SELECT table_name FROM information_schema.tables
             WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
-            AND table_name NOT IN (%s)""" % known_tables_repr)
+            AND table_name NOT IN %s""", (tuple(known_tables),))
 
         res = [(0, 0, {'name': row[0]}) for row in cr.fetchall()]
         if not res:

database_cleanup/model/purge_wizard.py

@@ -18,8 +18,9 @@
 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 ##############################################################################
-
 import logging
+from openerp import api, SUPERUSER_ID
+from openerp.exceptions import AccessDenied
 from openerp.osv import orm, fields
 
 
@@ -37,6 +38,15 @@ class CleanupPurgeLine(orm.AbstractModel):
     def purge(self, cr, uid, ids, context=None):
         raise NotImplementedError
 
+    @api.model
+    def create(self, values):
+        # make sure the user trying this is actually supposed to do it
+        if self.env.uid != SUPERUSER_ID and\
+           not self.env.ref('database_cleanup.menu_database_cleanup')\
+           .parent_id._filter_visible_menus():
+            raise AccessDenied
+        return super(CleanupPurgeLine, self).create(values)
+
 
 class PurgeWizard(orm.AbstractModel):
     """ Abstract base class for the purge wizards """
@@ -82,6 +92,15 @@ class PurgeWizard(orm.AbstractModel):
             'domain': [('wizard_id', 'in', ids)],
         }
 
+    @api.model
+    def create(self, values):
+        # make sure the user trying this is actually supposed to do it
+        if self.env.uid != SUPERUSER_ID and\
+           not self.env.ref('database_cleanup.menu_database_cleanup')\
+           .parent_id._filter_visible_menus():
+            raise AccessDenied
+        return super(PurgeWizard, self).create(values)
+
     _columns = {
         'name': fields.char('Name', size=64, readonly=True),
         }

database_cleanup/tests/__init__.py

@@ -0,0 +1,4 @@
+# -*- coding: utf-8 -*-
+# © 2016 Therp BV <http://therp.nl>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
+from . import test_database_cleanup

database_cleanup/tests/test_database_cleanup.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+# © 2016 Therp BV <http://therp.nl>
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
+from psycopg2 import ProgrammingError
+from openerp.tools import config
+from openerp.tests.common import TransactionCase
+
+
+class TestDatabaseCleanup(TransactionCase):
+    def test_database_cleanup(self):
+        # create an orphaned column
+        self.cr.execute(
+            'alter table res_users add column database_cleanup_test int')
+        purge_columns = self.env['cleanup.purge.wizard.column'].create({})
+        purge_columns.purge_all()
+        # must be removed by the wizard
+        with self.assertRaises(ProgrammingError):
+            with self.registry.cursor() as cr:
+                cr.execute('select database_cleanup_test from res_users')
+
+        # create a data entry pointing nowhere
+        self.cr.execute('select max(id) + 1 from res_users')
+        self.env['ir.model.data'].create({
+            'module': 'database_cleanup',
+            'name': 'test_no_data_entry',
+            'model': 'res.users',
+            'res_id': self.cr.fetchone()[0],
+        })
+        purge_data = self.env['cleanup.purge.wizard.data'].create({})
+        purge_data.purge_all()
+        # must be removed by the wizard
+        with self.assertRaises(ValueError):
+            self.env.ref('database_cleanup.test_no_data_entry')
+
+        # create a nonexistent model
+        self.env['ir.model'].create({
+            'name': 'Database cleanup test model',
+            'model': 'x_database.cleanup.test.model',
+        })
+        self.env.cr.execute(
+            'insert into ir_attachment (name, res_model, res_id, type) values '
+            "('test attachment', 'database.cleanup.test.model', 42, 'binary')")
+        self.registry.models.pop('x_database.cleanup.test.model')
+        self.registry._pure_function_fields.pop(
+            'x_database.cleanup.test.model')
+        purge_models = self.env['cleanup.purge.wizard.model'].create({})
+        purge_models.purge_all()
+        # must be removed by the wizard
+        self.assertFalse(self.env['ir.model'].search([
+            ('model', '=', 'x_database.cleanup.test.model'),
+        ]))
+
+        # create a nonexistent module
+        self.env['ir.module.module'].create({
+            'name': 'database_cleanup_test',
+            'state': 'to upgrade',
+        })
+        purge_modules = self.env['cleanup.purge.wizard.module'].create({})
+        # this reloads our registry, and we don't want to run tests twice
+        config.options['test_enable'] = False
+        purge_modules.purge_all()
+        config.options['test_enable'] = True
+        # must be removed by the wizard
+        self.assertFalse(self.env['ir.module.module'].search([
+            ('name', '=', 'database_cleanup_test'),
+        ]))
+
+        # create an orphaned table
+        self.env.cr.execute('create table database_cleanup_test (test int)')
+        purge_tables = self.env['cleanup.purge.wizard.table'].create({})
+        purge_tables.purge_all()
+        with self.assertRaises(ProgrammingError):
+            with self.registry.cursor() as cr:
+                self.env.cr.execute('select * from database_cleanup_test')