misc fixes

dbbackup/db/mongodb.py

@@ -8,46 +8,47 @@ class MongoDumpConnector(BaseCommandDBConnector):
     MongoDB connector, creates dump with ``mongodump`` and restore with
     ``mongorestore``.
     """
-    dump_cmd = 'mongodump'
-    restore_cmd = 'mongorestore'
+
+    dump_cmd = "mongodump"
+    restore_cmd = "mongorestore"
     object_check = True
     drop = True
 
     def _create_dump(self):
         cmd = f"{self.dump_cmd} --db {self.settings['NAME']}"
-        host = self.settings.get('HOST') or 'localhost'
-        port = self.settings.get('PORT') or 27017
-        cmd += f' --host {host}:{port}'
-        if self.settings.get('USER'):
+        host = self.settings.get("HOST") or "localhost"
+        port = self.settings.get("PORT") or 27017
+        cmd += f" --host {host}:{port}"
+        if self.settings.get("USER"):
             cmd += f" --username {self.settings['USER']}"
-        if self.settings.get('PASSWORD'):
+        if self.settings.get("PASSWORD"):
             cmd += f" --password {utils.get_escaped_command_arg(self.settings['PASSWORD'])}"
 
-        if self.settings.get('AUTH_SOURCE'):
+        if self.settings.get("AUTH_SOURCE"):
             cmd += f" --authenticationDatabase {self.settings['AUTH_SOURCE']}"
         for collection in self.exclude:
-            cmd += f' --excludeCollection {collection}'
-        cmd += ' --archive'
-        cmd = f'{self.dump_prefix} {cmd} {self.dump_suffix}'
-        stdout, stderr = self.run_command(cmd, env=self.dump_env)
+            cmd += f" --excludeCollection {collection}"
+        cmd += " --archive"
+        cmd = f"{self.dump_prefix} {cmd} {self.dump_suffix}"
+        stdout, _stderr = self.run_command(cmd, env=self.dump_env)
         return stdout
 
     def _restore_dump(self, dump):
         cmd = self.restore_cmd
-        host = self.settings.get('HOST') or 'localhost'
-        port = self.settings.get('PORT') or 27017
-        cmd += f' --host {host}:{port}'
-        if self.settings.get('USER'):
+        host = self.settings.get("HOST") or "localhost"
+        port = self.settings.get("PORT") or 27017
+        cmd += f" --host {host}:{port}"
+        if self.settings.get("USER"):
             cmd += f" --username {self.settings['USER']}"
-        if self.settings.get('PASSWORD'):
+        if self.settings.get("PASSWORD"):
             cmd += f" --password {utils.get_escaped_command_arg(self.settings['PASSWORD'])}"
 
-        if self.settings.get('AUTH_SOURCE'):
+        if self.settings.get("AUTH_SOURCE"):
             cmd += f" --authenticationDatabase {self.settings['AUTH_SOURCE']}"
         if self.object_check:
-            cmd += ' --objcheck'
+            cmd += " --objcheck"
         if self.drop:
-            cmd += ' --drop'
-        cmd += ' --archive'
-        cmd = f'{self.restore_prefix} {cmd} {self.restore_suffix}'
+            cmd += " --drop"
+        cmd += " --archive"
+        cmd = f"{self.restore_prefix} {cmd} {self.restore_suffix}"
         return self.run_command(cmd, stdin=dump, env=self.restore_env)

dbbackup/db/mysql.py

@@ -8,37 +8,38 @@ class MysqlDumpConnector(BaseCommandDBConnector):
     MySQL connector, creates dump with ``mysqldump`` and restore with
     ``mysql``.
     """
-    dump_cmd = 'mysqldump'
-    restore_cmd = 'mysql'
+
+    dump_cmd = "mysqldump"
+    restore_cmd = "mysql"
 
     def _create_dump(self):
         cmd = f"{self.dump_cmd} {self.settings['NAME']} --quick"
-        if self.settings.get('HOST'):
+        if self.settings.get("HOST"):
             cmd += f" --host={self.settings['HOST']}"
-        if self.settings.get('PORT'):
+        if self.settings.get("PORT"):
             cmd += f" --port={self.settings['PORT']}"
-        if self.settings.get('USER'):
+        if self.settings.get("USER"):
             cmd += f" --user={self.settings['USER']}"
-        if self.settings.get('PASSWORD'):
+        if self.settings.get("PASSWORD"):
             cmd += f" --password={utils.get_escaped_command_arg(self.settings['PASSWORD'])}"
 
         for table in self.exclude:
             cmd += f" --ignore-table={self.settings['NAME']}.{table}"
-        cmd = f'{self.dump_prefix} {cmd} {self.dump_suffix}'
-        stdout, stderr = self.run_command(cmd, env=self.dump_env)
+        cmd = f"{self.dump_prefix} {cmd} {self.dump_suffix}"
+        stdout, _stderr = self.run_command(cmd, env=self.dump_env)
         return stdout
 
     def _restore_dump(self, dump):
         cmd = f"{self.restore_cmd} {self.settings['NAME']}"
-        if self.settings.get('HOST'):
+        if self.settings.get("HOST"):
             cmd += f" --host={self.settings['HOST']}"
-        if self.settings.get('PORT'):
+        if self.settings.get("PORT"):
             cmd += f" --port={self.settings['PORT']}"
-        if self.settings.get('USER'):
+        if self.settings.get("USER"):
             cmd += f" --user={self.settings['USER']}"
-        if self.settings.get('PASSWORD'):
+        if self.settings.get("PASSWORD"):
             cmd += f" --password={utils.get_escaped_command_arg(self.settings['PASSWORD'])}"
 
-        cmd = f'{self.restore_prefix} {cmd} {self.restore_suffix}'
+        cmd = f"{self.restore_prefix} {cmd} {self.restore_suffix}"
         stdout, stderr = self.run_command(cmd, stdin=dump, env=self.restore_env)
         return stdout, stderr

dbbackup/db/postgresql.py

@@ -4,25 +4,25 @@
 from .base import BaseCommandDBConnector
 from .exceptions import DumpError
 
-logger = logging.getLogger('dbbackup.command')
+logger = logging.getLogger("dbbackup.command")
 
 
 def create_postgres_uri(self):
-    host = self.settings.get('HOST')
+    host = self.settings.get("HOST")
     if not host:
-        raise DumpError('A host name is required')
+        raise DumpError("A host name is required")
 
-    dbname = self.settings.get('NAME') or ''
-    user = quote(self.settings.get('USER') or '')
-    password = self.settings.get('PASSWORD') or ''
-    password = f':{quote(password)}' if password else ''
+    dbname = self.settings.get("NAME") or ""
+    user = quote(self.settings.get("USER") or "")
+    password = self.settings.get("PASSWORD") or ""
+    password = f":{quote(password)}" if password else ""
     if not user:
-        password = ''
+        password = ""
     else:
-        host = '@' + host
+        host = f"@{host}"
 
-    port = ':{}'.format(self.settings.get('PORT')) if self.settings.get('PORT') else ''
-    dbname = f'--dbname=postgresql://{user}{password}{host}{port}/{dbname}'
+    port = ":{}".format(self.settings.get("PORT")) if self.settings.get("PORT") else ""
+    dbname = f"--dbname=postgresql://{user}{password}{host}{port}/{dbname}"
     return dbname
 
 
@@ -31,35 +31,36 @@ class PgDumpConnector(BaseCommandDBConnector):
     PostgreSQL connector, it uses pg_dump`` to create an SQL text file
     and ``psql`` for restore it.
     """
-    extension = 'psql'
-    dump_cmd = 'pg_dump'
-    restore_cmd = 'psql'
+
+    extension = "psql"
+    dump_cmd = "pg_dump"
+    restore_cmd = "psql"
     single_transaction = True
     drop = True
 
     def _create_dump(self):
-        cmd = f'{self.dump_cmd} '
+        cmd = f"{self.dump_cmd} "
         cmd = cmd + create_postgres_uri(self)
 
         for table in self.exclude:
-            cmd += f' --exclude-table-data={table}'
+            cmd += f" --exclude-table-data={table}"
         if self.drop:
-            cmd += ' --clean'
+            cmd += " --clean"
 
-        cmd = f'{self.dump_prefix} {cmd} {self.dump_suffix}'
-        stdout, stderr = self.run_command(cmd, env=self.dump_env)
+        cmd = f"{self.dump_prefix} {cmd} {self.dump_suffix}"
+        stdout, _stderr = self.run_command(cmd, env=self.dump_env)
         return stdout
 
     def _restore_dump(self, dump):
-        cmd = f'{self.restore_cmd} '
+        cmd = f"{self.restore_cmd} "
         cmd = cmd + create_postgres_uri(self)
 
         # without this, psql terminates with an exit value of 0 regardless of errors
-        cmd += ' --set ON_ERROR_STOP=on'
+        cmd += " --set ON_ERROR_STOP=on"
         if self.single_transaction:
-            cmd += ' --single-transaction'
-        cmd += ' {}'.format(self.settings['NAME'])
-        cmd = f'{self.restore_prefix} {cmd} {self.restore_suffix}'
+            cmd += " --single-transaction"
+        cmd += " {}".format(self.settings["NAME"])
+        cmd = f"{self.restore_prefix} {cmd} {self.restore_suffix}"
         stdout, stderr = self.run_command(cmd, stdin=dump, env=self.restore_env)
         return stdout, stderr
 
@@ -69,21 +70,21 @@ class PgDumpGisConnector(PgDumpConnector):
     PostgreGIS connector, same than :class:`PgDumpGisConnector` but enable
     postgis if not made.
     """
-    psql_cmd = 'psql'
+
+    psql_cmd = "psql"
 
     def _enable_postgis(self):
-        cmd = '{} -c "CREATE EXTENSION IF NOT EXISTS postgis;"'.format(
-            self.psql_cmd)
-        cmd += ' --username={}'.format(self.settings['ADMIN_USER'])
-        cmd += ' --no-password'
-        if self.settings.get('HOST'):
-            cmd += ' --host={}'.format(self.settings['HOST'])
-        if self.settings.get('PORT'):
-            cmd += ' --port={}'.format(self.settings['PORT'])
+        cmd = '{} -c "CREATE EXTENSION IF NOT EXISTS postgis;"'.format(self.psql_cmd)
+        cmd += " --username={}".format(self.settings["ADMIN_USER"])
+        cmd += " --no-password"
+        if self.settings.get("HOST"):
+            cmd += " --host={}".format(self.settings["HOST"])
+        if self.settings.get("PORT"):
+            cmd += " --port={}".format(self.settings["PORT"])
         return self.run_command(cmd)
 
     def _restore_dump(self, dump):
-        if self.settings.get('ADMIN_USER'):
+        if self.settings.get("ADMIN_USER"):
             self._enable_postgis()
         return super()._restore_dump(dump)
 
@@ -93,31 +94,32 @@ class PgDumpBinaryConnector(PgDumpConnector):
     PostgreSQL connector, it uses pg_dump`` to create an SQL text file
     and ``pg_restore`` for restore it.
     """
-    extension = 'psql.bin'
-    dump_cmd = 'pg_dump'
-    restore_cmd = 'pg_restore'
+
+    extension = "psql.bin"
+    dump_cmd = "pg_dump"
+    restore_cmd = "pg_restore"
     single_transaction = True
     drop = True
 
     def _create_dump(self):
-        cmd = f'{self.dump_cmd} '
+        cmd = f"{self.dump_cmd} "
         cmd = cmd + create_postgres_uri(self)
 
-        cmd += ' --format=custom'
+        cmd += " --format=custom"
         for table in self.exclude:
-            cmd += f' --exclude-table-data={table}'
-        cmd = f'{self.dump_prefix} {cmd} {self.dump_suffix}'
-        stdout, stderr = self.run_command(cmd, env=self.dump_env)
+            cmd += f" --exclude-table-data={table}"
+        cmd = f"{self.dump_prefix} {cmd} {self.dump_suffix}"
+        stdout, _stderr = self.run_command(cmd, env=self.dump_env)
         return stdout
 
     def _restore_dump(self, dump):
         dbname = create_postgres_uri(self)
-        cmd = f'{self.restore_cmd} {dbname}'
+        cmd = f"{self.restore_cmd} {dbname}"
 
         if self.single_transaction:
-            cmd += ' --single-transaction'
+            cmd += " --single-transaction"
         if self.drop:
-            cmd += ' --clean'
-        cmd = f'{self.restore_prefix} {cmd} {self.restore_suffix}'
+            cmd += " --clean"
+        cmd = f"{self.restore_prefix} {cmd} {self.restore_suffix}"
         stdout, stderr = self.run_command(cmd, stdin=dump, env=self.restore_env)
         return stdout, stderr

dbbackup/tests/test_connectors/test_base.py

@@ -15,76 +15,76 @@ def test_get_connector(self):
 
 class BaseDBConnectorTest(TestCase):
     def test_init(self):
-        connector = BaseDBConnector()
+        BaseDBConnector()
 
     def test_settings(self):
         connector = BaseDBConnector()
-        connector.settings
+        connector.settings  # pylint: disable=pointless-statement
 
     def test_generate_filename(self):
         connector = BaseDBConnector()
-        filename = connector.generate_filename()
+        connector.generate_filename()
 
 
 class BaseCommandDBConnectorTest(TestCase):
     def test_run_command(self):
         connector = BaseCommandDBConnector()
-        stdout, stderr = connector.run_command('echo 123')
-        self.assertEqual(stdout.read(), b'123\n')
-        self.assertEqual(stderr.read(), b'')
+        stdout, _stderr = connector.run_command("echo 123")
+        self.assertEqual(stdout.read(), b"123\n")
+        self.assertEqual(stderr.read(), b"")
 
     def test_run_command_error(self):
         connector = BaseCommandDBConnector()
         with self.assertRaises(exceptions.CommandConnectorError):
-            connector.run_command('echa 123')
+            connector.run_command("echa 123")
 
     def test_run_command_stdin(self):
         connector = BaseCommandDBConnector()
         stdin = SpooledTemporaryFile()
-        stdin.write(b'foo')
+        stdin.write(b"foo")
         stdin.seek(0)
         # Run
-        stdout, stderr = connector.run_command('cat', stdin=stdin)
-        self.assertEqual(stdout.read(), b'foo')
+        stdout, _stderr = connector.run_command("cat", stdin=stdin)
+        self.assertEqual(stdout.read(), b"foo")
         self.assertFalse(stderr.read())
 
     def test_run_command_with_env(self):
         connector = BaseCommandDBConnector()
         # Empty env
-        stdout, stderr = connector.run_command('env')
+        stdout, _stderr = connector.run_command("env")
         self.assertTrue(stdout.read())
         # env from self.env
-        connector.env = {'foo': 'bar'}
-        stdout, stderr = connector.run_command('env')
-        self.assertIn(b'foo=bar\n', stdout.read())
+        connector.env = {"foo": "bar"}
+        stdout, _stderr = connector.run_command("env")
+        self.assertIn(b"foo=bar\n", stdout.read())
         # method overide gloabal env
-        stdout, stderr = connector.run_command('env', env={'foo': 'ham'})
-        self.assertIn(b'foo=ham\n', stdout.read())
+        stdout, _stderr = connector.run_command("env", env={"foo": "ham"})
+        self.assertIn(b"foo=ham\n", stdout.read())
         # get a var from parent env
-        os.environ['bar'] = 'foo'
-        stdout, stderr = connector.run_command('env')
-        self.assertIn(b'bar=foo\n', stdout.read())
+        os.environ["bar"] = "foo"
+        stdout, _stderr = connector.run_command("env")
+        self.assertIn(b"bar=foo\n", stdout.read())
         # Conf overides parendt env
-        connector.env = {'bar': 'bar'}
-        stdout, stderr = connector.run_command('env')
-        self.assertIn(b'bar=bar\n', stdout.read())
+        connector.env = {"bar": "bar"}
+        stdout, _stderr = connector.run_command("env")
+        self.assertIn(b"bar=bar\n", stdout.read())
         # method overides all
-        stdout, stderr = connector.run_command('env', env={'bar': 'ham'})
-        self.assertIn(b'bar=ham\n', stdout.read())
+        stdout, _stderr = connector.run_command("env", env={"bar": "ham"})
+        self.assertIn(b"bar=ham\n", stdout.read())
 
     def test_run_command_with_parent_env(self):
         connector = BaseCommandDBConnector(use_parent_env=False)
         # Empty env
-        stdout, stderr = connector.run_command('env')
+        stdout, _stderr = connector.run_command("env")
         self.assertFalse(stdout.read())
         # env from self.env
-        connector.env = {'foo': 'bar'}
-        stdout, stderr = connector.run_command('env')
-        self.assertEqual(stdout.read(), b'foo=bar\n')
+        connector.env = {"foo": "bar"}
+        stdout, _stderr = connector.run_command("env")
+        self.assertEqual(stdout.read(), b"foo=bar\n")
         # method overide gloabal env
-        stdout, stderr = connector.run_command('env', env={'foo': 'ham'})
-        self.assertEqual(stdout.read(), b'foo=ham\n')
+        stdout, _stderr = connector.run_command("env", env={"foo": "ham"})
+        self.assertEqual(stdout.read(), b"foo=ham\n")
         # no var from parent env
-        os.environ['bar'] = 'foo'
-        stdout, stderr = connector.run_command('env')
-        self.assertNotIn(b'bar=foo\n', stdout.read())
+        os.environ["bar"] = "foo"
+        stdout, _stderr = connector.run_command("env")
+        self.assertNotIn(b"bar=foo\n", stdout.read())