diff --git a/auto_backup/__manifest__.py b/auto_backup/__manifest__.py
index 801a2aa7364..6458377d51e 100644
--- a/auto_backup/__manifest__.py
+++ b/auto_backup/__manifest__.py
@@ -17,9 +17,7 @@
"license": "AGPL-3",
"website": "https://github.com/OCA/server-tools",
"category": "Tools",
- "depends": [
- "mail",
- ],
+ "depends": ["mail"],
"data": [
"data/ir_cron.xml",
"data/mail_message_subtype.xml",
@@ -27,7 +25,5 @@
"view/db_backup_view.xml",
],
"installable": True,
- "external_dependencies": {
- "python": ["pysftp"],
- },
+ "external_dependencies": {"python": ["pysftp"]},
}
diff --git a/auto_backup/data/ir_cron.xml b/auto_backup/data/ir_cron.xml
index 9fe1bfb76f4..a6d644cd7fe 100644
--- a/auto_backup/data/ir_cron.xml
+++ b/auto_backup/data/ir_cron.xml
@@ -1,16 +1,17 @@
-
+
-
Backup Scheduler
1
days
-1
-
-
+
+
code
model.action_backup_all()
-
diff --git a/auto_backup/data/mail_message_subtype.xml b/auto_backup/data/mail_message_subtype.xml
index a0e8e932bca..4f37c759803 100644
--- a/auto_backup/data/mail_message_subtype.xml
+++ b/auto_backup/data/mail_message_subtype.xml
@@ -1,18 +1,15 @@
-
+
-
Backup Successful
Database backup succeeded.
db.backup
-
Backup Failed
Database backup failed.
db.backup
-
diff --git a/auto_backup/models/db_backup.py b/auto_backup/models/db_backup.py
index be927eb7a04..561af770f1b 100644
--- a/auto_backup/models/db_backup.py
+++ b/auto_backup/models/db_backup.py
@@ -12,24 +12,28 @@
from glob import iglob
from odoo import _, api, exceptions, fields, models, tools
+from odoo.exceptions import UserError
from odoo.service import db
_logger = logging.getLogger(__name__)
try:
import pysftp
except ImportError: # pragma: no cover
- _logger.debug('Cannot import pysftp')
+ _logger.debug("Cannot import pysftp")
class DbBackup(models.Model):
- _description = 'Database Backup'
- _name = 'db.backup'
+ _description = "Database Backup"
+ _name = "db.backup"
_inherit = "mail.thread"
_sql_constraints = [
("name_unique", "UNIQUE(name)", "Cannot duplicate a configuration."),
- ("days_to_keep_positive", "CHECK(days_to_keep >= 0)",
- "I cannot remove backups from the future. Ask Doc for that."),
+ (
+ "days_to_keep_positive",
+ "CHECK(days_to_keep >= 0)",
+ "I cannot remove backups from the future. Ask Doc for that.",
+ ),
]
name = fields.Char(
@@ -39,14 +43,14 @@ class DbBackup(models.Model):
)
folder = fields.Char(
default=lambda self: self._default_folder(),
- help='Absolute path for storing the backups',
- required=True
+ help="Absolute path for storing the backups",
+ required=True,
)
days_to_keep = fields.Integer(
required=True,
default=0,
help="Backups older than this will be deleted automatically. "
- "Set 0 to disable autodeletion.",
+ "Set 0 to disable autodeletion.",
)
method = fields.Selection(
[("local", "Local disk"), ("sftp", "Remote SFTP server")],
@@ -54,53 +58,49 @@ class DbBackup(models.Model):
help="Choose the storage method for this backup.",
)
sftp_host = fields.Char(
- 'SFTP Server',
+ "SFTP Server",
help=(
"The host name or IP address from your remote"
" server. For example 192.168.0.1"
- )
+ ),
)
sftp_port = fields.Integer(
"SFTP Port",
default=22,
- help="The port on the FTP server that accepts SSH/SFTP calls."
+ help="The port on the FTP server that accepts SSH/SFTP calls.",
)
sftp_user = fields.Char(
- 'Username in the SFTP Server',
+ "Username in the SFTP Server",
help=(
"The username where the SFTP connection "
"should be made with. This is the user on the external server."
- )
+ ),
)
sftp_password = fields.Char(
"SFTP Password",
help="The password for the SFTP connection. If you specify a private "
- "key file, then this is the password to decrypt it.",
+ "key file, then this is the password to decrypt it.",
)
sftp_private_key = fields.Char(
"Private key location",
help="Path to the private key file. Only the Odoo user should have "
- "read permissions for that file.",
+ "read permissions for that file.",
)
backup_format = fields.Selection(
[
("zip", "zip (includes filestore)"),
- ("dump", "pg_dump custom format (without filestore)")
+ ("dump", "pg_dump custom format (without filestore)"),
],
- default='zip',
- help="Choose the format for this backup."
+ default="zip",
+ help="Choose the format for this backup.",
)
@api.model
def _default_folder(self):
"""Default to ``backups`` folder inside current server datadir."""
- return os.path.join(
- tools.config["data_dir"],
- "backups",
- self.env.cr.dbname)
+ return os.path.join(tools.config["data_dir"], "backups", self.env.cr.dbname)
- @api.multi
@api.depends("folder", "method", "sftp_host", "sftp_port", "sftp_user")
def _compute_name(self):
"""Get the right summary for this job."""
@@ -109,34 +109,40 @@ def _compute_name(self):
rec.name = "%s @ localhost" % rec.folder
elif rec.method == "sftp":
rec.name = "sftp://%s@%s:%d%s" % (
- rec.sftp_user, rec.sftp_host, rec.sftp_port, rec.folder)
+ rec.sftp_user,
+ rec.sftp_host,
+ rec.sftp_port,
+ rec.folder,
+ )
- @api.multi
@api.constrains("folder", "method")
def _check_folder(self):
"""Do not use the filestore or you will backup your backups."""
for record in self:
- if (record.method == "local" and
- record.folder.startswith(
- tools.config.filestore(self.env.cr.dbname))):
+ if record.method == "local" and record.folder.startswith(
+ tools.config.filestore(self.env.cr.dbname)
+ ):
raise exceptions.ValidationError(
- _("Do not save backups on your filestore, or you will "
- "backup your backups too!"))
+ _(
+ "Do not save backups on your filestore, or you will "
+ "backup your backups too!"
+ )
+ )
- @api.multi
def action_sftp_test_connection(self):
"""Check if the SFTP settings are correct."""
try:
# Just open and close the connection
with self.sftp_connection():
- raise exceptions.Warning(_("Connection Test Succeeded!"))
- except (pysftp.CredentialException,
- pysftp.ConnectionException,
- pysftp.SSHException):
+ raise UserError(_("Connection Test Succeeded!"))
+ except (
+ pysftp.CredentialException,
+ pysftp.ConnectionException,
+ pysftp.SSHException,
+ ) as exc:
_logger.info("Connection Test Failed!", exc_info=True)
- raise exceptions.Warning(_("Connection Test Failed!"))
+ raise UserError(_("Connection Test Failed!")) from exc
- @api.multi
def action_backup(self):
"""Run selected backups."""
backup = None
@@ -148,22 +154,19 @@ def action_backup(self):
with rec.backup_log():
# Directory must exist
try:
- os.makedirs(rec.folder)
- except OSError:
- pass
+ os.makedirs(rec.folder, exist_ok=True)
+ except OSError as exc:
+ _logger.exception("Action backup - OSError: %s" % exc)
- with open(os.path.join(rec.folder, filename),
- 'wb') as destiny:
+ with open(os.path.join(rec.folder, filename), "wb") as destiny:
# Copy the cached backup
- if backup:
+ if backup and backup == destiny.name:
with open(backup) as cached:
shutil.copyfileobj(cached, destiny)
# Generate new backup
else:
db.dump_db(
- self.env.cr.dbname,
- destiny,
- backup_format=rec.backup_format
+ self.env.cr.dbname, destiny, backup_format=rec.backup_format
)
backup = backup or destiny.name
successful |= rec
@@ -176,23 +179,23 @@ def action_backup(self):
with rec.backup_log():
cached = db.dump_db(
- self.env.cr.dbname,
- None,
- backup_format=rec.backup_format
+ self.env.cr.dbname, None, backup_format=rec.backup_format
)
with cached:
with rec.sftp_connection() as remote:
# Directory must exist
try:
- remote.makedirs(rec.folder)
- except pysftp.ConnectionException:
- pass
+ remote.makedirs(rec.folder, exist_ok=True)
+ except pysftp.ConnectionException as exc:
+ _logger.exception(
+ "pysftp ConnectionException: %s" % exc
+ )
# Copy cached backup to remote server
with remote.open(
- os.path.join(rec.folder, filename),
- "wb") as destiny:
+ os.path.join(rec.folder, filename), "wb"
+ ) as destiny:
shutil.copyfileobj(cached, destiny)
successful |= rec
@@ -204,7 +207,6 @@ def action_backup_all(self):
"""Run all scheduled backups."""
return self.search([]).action_backup()
- @api.multi
@contextmanager
def backup_log(self):
"""Log a backup result."""
@@ -215,63 +217,63 @@ def backup_log(self):
_logger.exception("Database backup failed: %s", self.name)
escaped_tb = tools.html_escape(traceback.format_exc())
self.message_post( # pylint: disable=translation-required
- body="
%s
%s
" % (
- _("Database backup failed."),
- escaped_tb),
- subtype=self.env.ref(
- "auto_backup.mail_message_subtype_failure"
- ),
+ body="%s
%s
"
+ % (_("Database backup failed."), escaped_tb),
+ subtype_id=self.env.ref("auto_backup.mail_message_subtype_failure").id,
)
else:
_logger.info("Database backup succeeded: %s", self.name)
self.message_post(body=_("Database backup succeeded."))
- @api.multi
def cleanup(self):
"""Clean up old backups."""
now = datetime.now()
for rec in self.filtered("days_to_keep"):
with rec.cleanup_log():
- oldest = self.filename(now - timedelta(days=rec.days_to_keep))
+ bu_format = rec.backup_format
+ file_extension = bu_format == "zip" and "dump.zip" or bu_format
+ oldest = self.filename(
+ now - timedelta(days=rec.days_to_keep), bu_format
+ )
if rec.method == "local":
- for name in iglob(os.path.join(rec.folder,
- "*.dump.zip")):
+ for name in iglob(
+ os.path.join(rec.folder, "*.%s" % file_extension)
+ ):
if os.path.basename(name) < oldest:
os.unlink(name)
elif rec.method == "sftp":
with rec.sftp_connection() as remote:
for name in remote.listdir(rec.folder):
- if (name.endswith(".dump.zip") and
- os.path.basename(name) < oldest):
- remote.unlink('%s/%s' % (rec.folder, name))
+ if (
+ name.endswith(".%s" % file_extension)
+ and os.path.basename(name) < oldest
+ ):
+ remote.unlink("{}/{}".format(rec.folder, name))
- @api.multi
@contextmanager
def cleanup_log(self):
"""Log a possible cleanup failure."""
self.ensure_one()
try:
_logger.info(
- "Starting cleanup process after database backup: %s",
- self.name)
+ "Starting cleanup process after database backup: %s", self.name
+ )
yield
except Exception:
_logger.exception("Cleanup of old database backups failed: %s")
escaped_tb = tools.html_escape(traceback.format_exc())
self.message_post( # pylint: disable=translation-required
- body="%s
%s
" % (
- _("Cleanup of old database backups failed."),
- escaped_tb),
- subtype=self.env.ref("auto_backup.failure"))
+ body="%s
%s
"
+ % (_("Cleanup of old database backups failed."), escaped_tb),
+ subtype_id=self.env.ref("auto_backup.failure").id,
+ )
else:
- _logger.info(
- "Cleanup of old database backups succeeded: %s",
- self.name)
+ _logger.info("Cleanup of old database backups succeeded: %s", self.name)
@staticmethod
- def filename(when, ext='zip'):
+ def filename(when, ext="zip"):
"""Generate a file name for a backup.
:param datetime.datetime when:
@@ -279,10 +281,9 @@ def filename(when, ext='zip'):
:param str ext: Extension of the file. Default: dump.zip
"""
return "{:%Y_%m_%d_%H_%M_%S}.{ext}".format(
- when, ext='dump.zip' if ext == 'zip' else ext
+ when, ext="dump.zip" if ext == "zip" else ext
)
- @api.multi
def sftp_connection(self):
"""Return a new SFTP connection with found parameters."""
self.ensure_one()
@@ -292,8 +293,8 @@ def sftp_connection(self):
"port": self.sftp_port,
}
_logger.debug(
- "Trying to connect to sftp://%(username)s@%(host)s:%(port)d",
- extra=params)
+ "Trying to connect to sftp://%(username)s@%(host)s:%(port)d", extra=params
+ )
if self.sftp_private_key:
params["private_key"] = self.sftp_private_key
if self.sftp_password:
diff --git a/auto_backup/tests/test_db_backup.py b/auto_backup/tests/test_db_backup.py
index a30fa9c3311..f0df8c34643 100644
--- a/auto_backup/tests/test_db_backup.py
+++ b/auto_backup/tests/test_db_backup.py
@@ -98,6 +98,16 @@ def test_check_folder(self):
),
})
+ def test_check_folder_exists(self):
+ """ It should log about already existing folder """
+ rec_id = self.new_record('local')
+ rec_id.write({
+ 'folder': rec_id.folder,
+ })
+ rec_id.action_backup()
+ # No error was raised, just a log, test pass
+ self.assertTrue(True)
+
@mock.patch('%s._' % model)
def test_action_sftp_test_connection_success(self, _):
""" It should raise connection succeeded warning """
@@ -150,7 +160,26 @@ def test_action_backup_sftp_mkdirs(self):
with self.patch_filtered_sftp(rec_id):
conn = rec_id.sftp_connection().__enter__()
rec_id.action_backup()
- conn.makedirs.assert_called_once_with(rec_id.folder)
+ conn.makedirs.assert_called_once_with(
+ rec_id.folder, exist_ok=True
+ )
+
+ def test_action_backup_sftp_cleanup(self):
+ """ Backup sftp database and cleanup old databases """
+ rec_id = self.new_record()
+ with self.mock_assets():
+ with self.patch_filtered_sftp(rec_id):
+ conn = rec_id.sftp_connection().__enter__()
+ rec_id.days_to_keep = 1
+ filename = rec_id.filename(
+ datetime.now(), ext=rec_id.backup_format
+ )
+ rec_id.action_backup()
+ rec_id.days_to_keep = 0
+ rec_id.action_backup()
+ conn.unlink.assert_called_once_with(
+ "{}/{}".format(rec_id.folder, filename)
+ )
def test_action_backup_sftp_mkdirs_conn_exception(self):
""" It should guard from ConnectionException on remote.mkdirs """
diff --git a/auto_backup/view/db_backup_view.xml b/auto_backup/view/db_backup_view.xml
index 9b426e5b836..d4dc9a95168 100644
--- a/auto_backup/view/db_backup_view.xml
+++ b/auto_backup/view/db_backup_view.xml
@@ -1,21 +1,27 @@
-
+
-
db.backup
-
db.backup
-
-
-
+
+
+
-
db.backup
-
-
-
+
+
+
-
-
-
+
+ Automated Backups
+ db.backup
+
-
+ id="backup_conf_menu"
+ />
Execute backup(s)
@@ -91,5 +97,4 @@
code
records.action_backup()
-