VMM/domain: Simplified Domain.delete() related methods and their arguments, v0.6.x
authorPascal Volk <neverseen@users.sourceforge.net>
Fri, 30 Jul 2010 15:19:31 +0000
branchv0.6.x
changeset 331 270b57af85de
parent 330 3fe8d6cdbe3a
child 332 352ca7f1b332
VMM/domain: Simplified Domain.delete() related methods and their arguments, VMM/handler: Adjusted to the changes in the domain module.
VirtualMailManager/domain.py
VirtualMailManager/handler.py
--- a/VirtualMailManager/domain.py	Fri Jul 30 09:45:51 2010 +0000
+++ b/VirtualMailManager/domain.py	Fri Jul 30 15:19:31 2010 +0000
@@ -13,10 +13,10 @@
 from random import choice
 
 from VirtualMailManager.constants import \
-     ACCOUNT_AND_ALIAS_PRESENT, ACCOUNT_PRESENT, ALIAS_PRESENT, \
-     DOMAIN_ALIAS_EXISTS, DOMAIN_EXISTS, DOMAIN_INVALID, DOMAIN_TOO_LONG, \
-     NO_SUCH_DOMAIN
+     ACCOUNT_AND_ALIAS_PRESENT, DOMAIN_ALIAS_EXISTS, DOMAIN_EXISTS, \
+     DOMAIN_INVALID, DOMAIN_TOO_LONG, NO_SUCH_DOMAIN
 from VirtualMailManager.errors import DomainError as DomErr
+from VirtualMailManager.pycompat import any
 from VirtualMailManager.transport import Transport
 
 
@@ -83,49 +83,24 @@
         self._gid = dbc.fetchone()[0]
         dbc.close()
 
-    def _has(self, what):
-        """Checks if aliases or accounts are assigned to the domain.
-
-        If there are assigned accounts or aliases True will be returned,
-        otherwise False will be returned.
-
-        Argument:
-
-        `what` : basestring
-            "alias" or "users"
+    def _check_for_addresses(self):
+        """Checks dependencies for deletion. Raises a DomainError if there
+        are accounts, aliases and/or relocated users.
         """
-        assert what in ('alias', 'users')
         dbc = self._dbh.cursor()
-        if what == 'users':
-            dbc.execute("SELECT count(gid) FROM users WHERE gid=%s", self._gid)
-        else:
-            dbc.execute("SELECT count(gid) FROM alias WHERE gid=%s", self._gid)
-        count = dbc.fetchone()
+        dbc.execute('SELECT count(gid) FROM users WHERE gid = %(gid)u '
+                    'UNION SELECT count(gid) FROM alias WHERE gid = %(gid)u '
+                    'UNION SELECT count(gid) FROM relocated WHERE gid = '
+                    '%(gid)u' % {'gid': self._gid})
+        result = dbc.fetchall()
         dbc.close()
-        return count[0] > 0
-
-    def _chk_delete(self, deluser, delalias):
-        """Checks dependencies for deletion.
-
-        Arguments:
-        deluser -- ignore available accounts (bool)
-        delalias -- ignore available aliases (bool)
-        """
-        if not deluser:
-            hasuser = self._has('users')
-        else:
-            hasuser = False
-        if not delalias:
-            hasalias = self._has('alias')
-        else:
-            hasalias = False
-        if hasuser and hasalias:
-            raise DomErr(_(u'There are accounts and aliases.'),
+        result = [count[0] for count in result]
+        if any(result):
+            keys = ('account_count', 'alias_count', 'relocated_count')
+            raise DomErr(_(u'There are %(account_count)u accounts, '
+                           u'%(alias_count)u aliases and %(relocated_count)u '
+                           u'relocated users.') % dict(zip(keys, result)),
                          ACCOUNT_AND_ALIAS_PRESENT)
-        elif hasuser:
-            raise DomErr(_(u'There are accounts.'), ACCOUNT_PRESENT)
-        elif hasalias:
-            raise DomErr(_(u'There are aliases.'), ALIAS_PRESENT)
 
     def _chk_state(self):
         """Throws a DomainError if the Domain is new - not saved in the
@@ -193,22 +168,26 @@
         dbc.close()
         self._new = False
 
-    def delete(self, deluser=False, delalias=False):
+    def delete(self, force=False):
         """Deletes the domain.
 
         Arguments:
 
-        `deluser` : bool
-          force deletion of all available accounts, default `False`
-        `delalias` : bool
-          force deletion of all available aliases, default `False`
+        `force` : bool
+          force the deletion of all available accounts, aliases and
+          relocated users.  When *force* is `False` and there are accounts,
+          aliases and/or relocated users a DomainError will be raised.
+          Default `False`
         """
+        if not isinstance(force, bool):
+            raise TypeError('force must be a bool')
         self._chk_state()
-        self._chk_delete(deluser, delalias)
+        if not force:
+            self._check_for_addresses()
         dbc = self._dbh.cursor()
         for tbl in ('alias', 'users', 'relocated', 'domain_name',
                     'domain_data'):
-            dbc.execute("DELETE FROM %s WHERE gid = %d" % (tbl, self._gid))
+            dbc.execute("DELETE FROM %s WHERE gid = %u" % (tbl, self._gid))
         self._dbh.commit()
         dbc.close()
         self._gid = 0
--- a/VirtualMailManager/handler.py	Fri Jul 30 09:45:51 2010 +0000
+++ b/VirtualMailManager/handler.py	Fri Jul 30 15:19:31 2010 +0000
@@ -389,22 +389,17 @@
         else:
             dom.update_transport(trsp, force=True)
 
-    def domain_delete(self, domainname, force=None):
+    def domain_delete(self, domainname, force=False):
         """Wrapper around Domain.delete()"""
-        if force and force not in ('deluser', 'delalias', 'delall'):
-            raise DomainError(_(u"Invalid argument: '%s'") % force,
-                              INVALID_ARGUMENT)
+        if not isinstance(force, bool):
+            raise TypeError('force must be a bool')
         dom = self._get_domain(domainname)
         gid = dom.gid
         domdir = dom.directory
-        if self._cfg.dget('domain.force_deletion') or force == 'delall':
-            dom.delete(True, True)
-        elif force == 'deluser':
-            dom.delete(deluser=True)
-        elif force == 'delalias':
-            dom.delete(delalias=True)
+        if self._cfg.dget('domain.force_deletion') or force:
+            dom.delete(True)
         else:
-            dom.delete()
+            dom.delete(False)
         if self._cfg.dget('domain.delete_directory'):
             self._delete_domain_dir(domdir, gid)