Skip to content

Commit

Permalink
Code cleanup
Browse files Browse the repository at this point in the history
Signed-off-by: Volker Theile <votdev@gmx.de>
  • Loading branch information
votdev committed Nov 4, 2018
1 parent cc1dee7 commit f307c90
Show file tree
Hide file tree
Showing 5 changed files with 137 additions and 13 deletions.
Expand Up @@ -53,6 +53,12 @@ class CommandHelper:
_backup_path = None

def mkBackup(self):
"""
.. deprecated:: 5.0
"""
return self.create_backup()

def create_backup(self):
"""
Create a backup of the configuration database.
:returns: Returns the path of the backup file, otherwise None.
Expand All @@ -66,6 +72,12 @@ def mkBackup(self):
return self._backup_path

def unlinkBackup(self):
"""
.. deprecated:: 5.0
"""
return self.unlink_backup()

def unlink_backup(self):
"""
Unlink the backup of the configuration database.
"""
Expand All @@ -77,6 +89,12 @@ def unlinkBackup(self):
self._backup_path = None

def rollbackChanges(self):
"""
.. deprecated:: 5.0
"""
return self.rollback_changes()

def rollback_changes(self):
"""
Rollback all changes in the configuration database.
"""
Expand All @@ -92,7 +110,7 @@ def argparse_is_uuid4(self, arg):
"""
Check if the specified value is a valid UUID4.
:param arg: The value to check.
:returns: The specified value.
:returns: The specified value.
:raises argparse.ArgumentTypeError:
"""
if not openmediavault.string.is_uuid4(arg):
Expand All @@ -103,7 +121,7 @@ def argparse_is_json(self, arg):
"""
Check if the specified value is a valid JSON string.
:param arg: The value to check.
:returns: The specified value as Python dictionary.
:returns: The specified value as Python dictionary.
:raises argparse.ArgumentTypeError:
"""
if not openmediavault.string.is_json(arg):
Expand All @@ -115,7 +133,7 @@ def argparse_is_json_stdin(self, arg):
Check if the specified value is a valid JSON string. Loads the
data from STDIN if '-' is given.
:param arg: The value to check.
:returns: The specified value as Python dictionary.
:returns: The specified value as Python dictionary.
:raises argparse.ArgumentTypeError:
"""
if arg == "-":
Expand All @@ -127,7 +145,7 @@ def argparse_is_datamodel_id(self, arg):
Check if the specified value is a valid datamodel ID.
Example: conf.service.ftp
:param arg: The value to check.
:returns: The specified value.
:returns: The specified value.
:raises argparse.ArgumentTypeError:
"""
if not re.match(r'^conf(\..+)?$', arg):
Expand Down
Expand Up @@ -62,7 +62,7 @@ def execute(self, *args):
break
try:
# Create a backup of the configuration database.
self.mkBackup()
self.create_backup()
# Test if the script exists and is executable.
script_path = os.path.join(create_dir, script_name)
if not os.path.exists(script_path):
Expand All @@ -84,10 +84,10 @@ def execute(self, *args):
"configuration: %s" % str(e)
)
# Rollback all changes.
self.rollbackChanges()
self.rollback_changes()
finally:
# Unlink the configuration database backup.
self.unlinkBackup()
self.unlink_backup()
return rc


Expand Down
Expand Up @@ -49,7 +49,7 @@ def execute(self, *args):
group.add_argument("--filter", nargs="?", type=self.argparse_is_json)
cmd_args = parser.parse_args(args[2:])
# Create a backup of the configuration database.
self.mkBackup()
self.create_backup()
# Get the database.
db = openmediavault.config.Database()
try:
Expand Down Expand Up @@ -78,10 +78,10 @@ def execute(self, *args):
"configuration object: %s" % str(e)
)
# Rollback all changes.
self.rollbackChanges()
self.rollback_changes()
finally:
# Unlink the configuration database backup.
self.unlinkBackup()
self.unlink_backup()
# Output the number of deleted configuration objects.
if 0 == rc:
print("Deleted %d object(s)" % len(objs))
Expand Down
Expand Up @@ -78,7 +78,7 @@ def execute(self, *args):
migrations[parts[1]] = name
try:
# Create a backup of the configuration database.
self.mkBackup()
self.create_backup()
# Execute the configuration database migration scripts.
for cmd_args.version in sorted(
migrations, key=lambda v: LooseVersion(v)
Expand All @@ -101,10 +101,10 @@ def execute(self, *args):
# Display the exception message.
openmediavault.log.error("Failed to apply migrations: %s" % str(e))
# Rollback all changes.
self.rollbackChanges()
self.rollback_changes()
finally:
# Unlink the configuration database backup.
self.unlinkBackup()
self.unlink_backup()
return rc


Expand Down
@@ -0,0 +1,106 @@
# -*- coding: utf-8 -*-
#
# This file is part of OpenMediaVault.
#
# @license http://www.gnu.org/licenses/gpl.html GPL Version 3
# @author Volker Theile <volker.theile@openmediavault.org>
# @copyright Copyright (c) 2009-2018 Volker Theile
#
# OpenMediaVault is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# any later version.
#
# OpenMediaVault is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with OpenMediaVault. If not, see <http://www.gnu.org/licenses/>.
import unittest
import mock
import openmediavault.confdbadm


class ConfDbAdmTestCase(unittest.TestCase):
def setUp(self):
self.command_helper = openmediavault.confdbadm.CommandHelper()

def test_create_backup_fail(self):
openmediavault.setenv("OMV_CONFIG_FILE", "xyz.conf")
self.assertIsNone(self.command_helper.create_backup())
self.assertFalse(self.command_helper._backup_path)

@mock.patch("os.path.exists", return_value=True)
@mock.patch("shutil.copy")
def test_create_backup(self, mock_copy, mock_exists):
openmediavault.setenv("OMV_CONFIG_FILE", "abc.conf")
result = self.command_helper.create_backup()
self.assertIsNotNone(result)
self.assertIsNotNone(self.command_helper._backup_path)
self.assertEqual(result, self.command_helper._backup_path)
mock_copy.assert_called_with("abc.conf", result)

def test_unlink_backup_fail(self):
with self.assertRaises(Exception) as ctx:
self.command_helper.unlink_backup()
self.assertEqual(
str(ctx.exception), "No configuration backup exists."
)

@mock.patch("os.unlink")
def test_unlink_backup_fail_2(self, unlink_mock):
self.command_helper._backup_path = False
self.command_helper.unlink_backup()
unlink_mock.assert_not_called()

@mock.patch("os.unlink")
def test_unlink_backup(self, unlink_mock):
self.command_helper._backup_path = "foo.conf"
self.command_helper.unlink_backup()
unlink_mock.assert_called_once()
self.assertIsNone(self.command_helper._backup_path)

def test_rollback_changes_fail(self):
with self.assertRaises(Exception) as ctx:
self.command_helper.rollback_changes()
self.assertEqual(
str(ctx.exception), "No configuration backup exists."
)

@mock.patch("shutil.copy")
def test_rollback_changes_fail_2(self, mock_copy):
self.command_helper._backup_path = False
self.command_helper.rollback_changes()
mock_copy.assert_not_called()

@mock.patch("shutil.copy")
def test_rollback_changes(self, mock_copy):
openmediavault.setenv("OMV_CONFIG_FILE", "xyz.conf")
self.command_helper._backup_path = "bar.conf"
self.command_helper.rollback_changes()
mock_copy.assert_called_with("bar.conf", "xyz.conf")

def test_argparse_is_uuid4(self):
uuid = "e4a987ee-dfd1-11e8-bae1-2321e792b9bd"
self.assertEqual(uuid, self.command_helper.argparse_is_uuid4(uuid))

def test_argparse_is_uuid4_fail(self):
with self.assertRaises(Exception) as ctx:
self.command_helper.argparse_is_uuid4("xyz")
self.assertEqual(str(ctx.exception), "No valid UUID4.")

def test_argparse_is_json(self):
self.assertDictEqual({
"a": 0
}, self.command_helper.argparse_is_json('{"a": 0}'))

def test_argparse_is_json_fail(self):
with self.assertRaises(Exception) as ctx:
self.command_helper.argparse_is_json("abc")
self.assertEqual(str(ctx.exception), "No valid JSON.")


if __name__ == "__main__":
unittest.main()

0 comments on commit f307c90

Please sign in to comment.