Add unit tests for edge modules. Add RAID level enumerator (#23)

* * Add RAIDLevel enumerator to edge/enum
* Update edge/array param documentation to RAIDLevel instead of string
* Remove redundant code from edge/telnet
* Update ut/base assert_equal_objects() to support deep comparison of objects
* Add form_data to ut/base_edge mock
* Add unit tests for modules: array, backup, cli, drive, groups, login, mail, network, power, shell, sync (cloudsync), telnet and timezome
* Add test for active directory connection error
* Add test for disabling mail server
* Add test for for modifying users
* Add tests for shares

cterasdk/edge/array.py

@@ -20,7 +20,7 @@ def add(self, array_name, level, members):
         Add a new array
 
         :param str array_name: Name for the new array
-        :param str level:
+        :param RAIDLevel level: RAID level
         :param list(str) members: Members of the array
         """
         param = Object()

cterasdk/edge/enum.py

@@ -301,3 +301,20 @@ class ServicesConnectionState:
     """
     Disconnected = "Disconnected"
     Connected = "Connected"
+
+
+class RAIDLevel:
+    """
+    RAID Levels
+
+    :ivar str JBOD: Linear concatenation
+    :ivar str RAID_0: Stripe set
+    :ivar str RAID_1: Mirror
+    :ivar str RAID_5: Distributed parity
+    :ivar str RAID_6: Dual parity
+    """
+    JBOD = "linear"
+    RAID_0 = "0"
+    RAID_1 = "1"
+    RAID_5 = "5"
+    RAID_6 = "6"

cterasdk/edge/telnet.py

@@ -10,11 +10,6 @@ class Telnet(BaseCommand):
 
     def enable(self, code):
         """ Enable Telnet """
-        status = self._gateway.get('/status/device')
-        version = status.runningFirmware
-        if version.count('.') == 2:
-            version = version + '.0'
-
         param = Object()
         param.code = code
 

tests/ut/base.py

@@ -1,5 +1,8 @@
+import queue
 import unittest
 
+from cterasdk.common import Object
+
 
 class BaseTest(unittest.TestCase):
 
@@ -15,5 +18,13 @@ def patch_property(self, module_path, **patch_kwargs):
         return self.patch_call(module_path, **patch_kwargs)
 
     def _assert_equal_objects(self, expected_param, actual_param):
-        for field in [a for a in dir(actual_param) if not a.startswith('__')]:
-            self.assertEqual(getattr(actual_param, field), getattr(expected_param, field))
+        q = queue.Queue()
+        q.put((actual_param, expected_param))
+        while not q.empty():
+            actual_param, expected_param = q.get()
+            for field in [a for a in dir(actual_param) if not a.startswith('__')]:
+                actual_param_attr = getattr(actual_param, field)
+                if isinstance(actual_param_attr, Object):
+                    q.put((actual_param_attr, getattr(expected_param, field)))
+                else:
+                    self.assertEqual(actual_param_attr, getattr(expected_param, field))

tests/ut/base_edge.py

@@ -11,10 +11,12 @@ def setUp(self):
         self._filer = Gateway("")
 
     def _init_filer(self, get_response=None, put_response=None, post_response=None,
-                    add_response=None, execute_response=None, delete_response=None):
+                    form_data_response=None, add_response=None, execute_response=None,
+                    delete_response=None):
         self._filer.get = mock.MagicMock(return_value=get_response)
         self._filer.put = mock.MagicMock(return_value=put_response)
         self._filer.post = mock.MagicMock(return_value=post_response)
+        self._filer.form_data = mock.MagicMock(return_value=form_data_response)
         self._filer.add = mock.MagicMock(return_value=add_response)
         self._filer.execute = mock.MagicMock(return_value=execute_response)
         self._filer.delete = mock.MagicMock(return_value=delete_response)

tests/ut/test_edge_array.py

@@ -0,0 +1,83 @@
+from unittest import mock
+
+from cterasdk import exception
+from cterasdk.edge import array
+from cterasdk.edge.enum import RAIDLevel
+from cterasdk.common import Object
+from tests.ut import base_edge
+
+
+class TestEdgeArray(base_edge.BaseEdgeTest):
+
+    def setUp(self):
+        super().setUp()
+        self._array_name = 'array'
+        self._array_level = RAIDLevel.JBOD
+        self._array_members = ['SATA-%s' % i for i in range(1, 9)]
+
+    def test_get_all_arrays(self):
+        get_response = 'Success'
+        self._init_filer(get_response=get_response)
+        ret = array.Array(self._filer).get()
+        self._filer.get.assert_called_once_with('/config/storage/arrays')
+        self.assertEqual(ret, get_response)
+
+    def test_get_array(self):
+        get_response = 'Success'
+        self._init_filer(get_response=get_response)
+        ret = array.Array(self._filer).get(self._array_name)
+        self._filer.get.assert_called_once_with('/config/storage/arrays/' + self._array_name)
+        self.assertEqual(ret, get_response)
+
+    def test_add_array_success(self):
+        add_response = 'Success'
+        self._init_filer(add_response=add_response)
+        ret = array.Array(self._filer).add(self._array_name, self._array_level, self._array_members)
+        self._filer.add.assert_called_once_with('/config/storage/arrays', mock.ANY)
+
+        expected_param = self._get_array_object()
+        actual_param = self._filer.add.call_args[0][1]
+        self._assert_equal_objects(expected_param, actual_param)
+
+        self.assertEqual(ret, add_response)
+
+    def test_add_array_failure(self):
+        expected_exception = exception.CTERAException()
+        self._filer.add = mock.MagicMock(side_effect=expected_exception)
+        with self.assertRaises(exception.CTERAException) as error:
+            array.Array(self._filer).add(self._array_name, self._array_level, self._array_members)
+        self.assertEqual('Storage array creation failed.', error.exception.message)
+
+    def test_delete_array_success(self):
+        delete_response = 'Success'
+        self._init_filer(delete_response=delete_response)
+        ret = array.Array(self._filer).delete(self._array_name)
+        self._filer.delete.assert_called_once_with('/config/storage/arrays/' + self._array_name)
+        self.assertEqual(ret, delete_response)
+
+    def test_delete_array_failure(self):
+        expected_exception = exception.CTERAException()
+        self._filer.delete = mock.MagicMock(side_effect=expected_exception)
+        with self.assertRaises(exception.CTERAException) as error:
+            array.Array(self._filer).delete(self._array_name)
+        self.assertEqual('Storage array deletion failed.', error.exception.message)
+
+    def test_delete_all(self):
+        self.patch_call("cterasdk.edge.array.Array.delete")
+        arrays = self._get_arrays_param()
+        self._init_filer(get_response=arrays)
+        array.Array(self._filer).delete_all()
+        self._filer.get.assert_called_once_with('/config/storage/arrays')
+
+    def _get_array_object(self):
+        array_param = Object()
+        array_param.name = self._array_name
+        array_param.level = self._array_level
+        array_param.members = self._array_members
+        return array_param
+
+    def _get_arrays_param(self):
+        array_param = Object()
+        array_param.name = self._array_name
+        arrays = [array_param]
+        return arrays

tests/ut/test_edge_backup.py

@@ -0,0 +1,26 @@
+from cterasdk.edge import backup
+from tests.ut import base_edge
+
+
+class TestEdgeBackup(base_edge.BaseEdgeTest):
+
+    def setUp(self):
+        super().setUp()
+        self._task_id = '137'
+        self._shared_secret = 'shared secret'
+        self._passphrase_salt = 'salt'
+
+    def test_start_backup(self):
+        self._init_filer()
+        backup.Backup(self._filer).start()
+        self._filer.execute.assert_called_once_with('/status/sync', 'start')
+
+    def test_suspend_backup(self):
+        self._init_filer()
+        backup.Backup(self._filer).suspend()
+        self._filer.execute.assert_called_once_with('/status/sync', 'pause')
+
+    def test_unsuspend_backup(self):
+        self._init_filer()
+        backup.Backup(self._filer).unsuspend()
+        self._filer.execute.assert_called_once_with('/status/sync', 'resume')

tests/ut/test_edge_cli.py

@@ -0,0 +1,16 @@
+from cterasdk.edge import cli
+from tests.ut import base_edge
+
+
+class TestEdgeCLI(base_edge.BaseEdgeTest):
+
+    def setUp(self):
+        super().setUp()
+        self._cli_command = 'show /config/fileservices/cifs'
+
+    def test_run_cli_command(self):
+        execute_response = 'Success'
+        self._init_filer(execute_response=execute_response)
+        ret = cli.CLI(self._filer).run_command(self._cli_command)
+        self._filer.execute.assert_called_once_with('/config/device', 'debugCmd', self._cli_command)
+        self.assertEqual(ret, execute_response)

tests/ut/test_edge_directory_service.py

@@ -17,12 +17,15 @@ def setUp(self):
         self._domain_flat_name = "CTERA"
         self._mapping_min = 2000000
         self._mapping_max = 5000000
-        self._filer.network.tcp_connect = mock.MagicMock()
 
     def test_connect(self):
         get_response = self._get_workgroup_param()
         self._init_filer(get_response=get_response)
+        tcp_connect_return_value = True
+        self._filer.network.tcp_connect = mock.MagicMock(return_value=tcp_connect_return_value)
+
         directoryservice.DirectoryService(self._filer).connect(self._domain, self._username, self._password)
+
         self._filer.network.tcp_connect.assert_called_once_with(address=self._domain, port=389)
         self._filer.get.assert_called_once_with('/config/fileservices/cifs')
         self._filer.put.assert_called_once_with('/config/fileservices/cifs', mock.ANY)
@@ -40,7 +43,11 @@ def test_connect_with_ou_path(self):
         ou_path = "ou=North America,DC=ctera,DC=local"
         get_response = self._get_workgroup_param()
         self._init_filer(get_response=get_response)
+        tcp_connect_return_value = True
+        self._filer.network.tcp_connect = mock.MagicMock(return_value=tcp_connect_return_value)
+
         directoryservice.DirectoryService(self._filer).connect(self._domain, self._username, self._password, ou_path)
+
         self._filer.network.tcp_connect.assert_called_once_with(address=self._domain, port=389)
         self._filer.get.assert_called_once_with('/config/fileservices/cifs')
         self._filer.put.assert_called_once_with('/config/fileservices/cifs', mock.ANY)
@@ -63,6 +70,16 @@ def test_connect_raise(self):
         with self.assertRaises(exception.CTERAException):
             directoryservice.DirectoryService(self._filer).connect(self._domain, self._username, self._password)
 
+    def test_connect_connection_error(self):
+        tcp_connect_return_value = False
+        self._filer.network.tcp_connect = mock.MagicMock(return_value=tcp_connect_return_value)
+
+        with self.assertRaises(exception.CTERAConnectionError) as error:
+            directoryservice.DirectoryService(self._filer).connect(self._domain, self._username, self._password)
+
+        self._filer.network.tcp_connect.assert_called_once_with(address=self._domain, port=389)
+        self.assertEqual('Unable to establish connection', error.exception.message)
+
     def test_set_advanced_mapping(self):
         get_response = TestEdgeDirectoryService._get_advanced_mapping_object(self._domain_flat_name, 0, 0)
         self._init_filer(get_response=get_response)

tests/ut/test_edge_drive.py

@@ -0,0 +1,52 @@
+from unittest import mock
+
+from cterasdk.edge import drive
+from cterasdk.common import Object
+from tests.ut import base_edge
+
+
+class TestEdgeDrive(base_edge.BaseEdgeTest):
+
+    def setUp(self):
+        super().setUp()
+        self._drive_name = 'XVD2'
+
+    def test_get_all_drives(self):
+        get_response = 'Success'
+        self._init_filer(get_response=get_response)
+        ret = drive.Drive(self._filer).get()
+        self._filer.get.assert_called_once_with('/config/storage/disks')
+        self.assertEqual(ret, get_response)
+
+    def test_get_drive(self):
+        get_response = 'Success'
+        self._init_filer(get_response=get_response)
+        ret = drive.Drive(self._filer).get(self._drive_name)
+        self._filer.get.assert_called_once_with('/config/storage/disks/' + self._drive_name)
+        self.assertEqual(ret, get_response)
+
+    def test_format_drive(self):
+        self._init_filer()
+        drive.Drive(self._filer).format(self._drive_name)
+        self._filer.execute.assert_called_once_with('/proc/storage', 'format', mock.ANY)
+
+        expected_param = Object()
+        expected_param.name = self._drive_name
+        actual_param = self._filer.execute.call_args[0][2]
+        self._assert_equal_objects(expected_param, actual_param)
+
+    def test_format_all_drives(self):
+        self._init_filer(get_response=self._get_drives_param())
+        drive.Drive(self._filer).format_all()
+        self._filer.get.assert_called_once_with('/status/storage/disks')
+
+        expected_param = Object()
+        expected_param.name = self._drive_name
+        actual_param = self._filer.execute.call_args[0][2]
+        self._assert_equal_objects(expected_param, actual_param)
+
+    def _get_drives_param(self):
+        drive_param = Object()
+        drive_param.name = self._drive_name
+        drives = [drive_param]
+        return drives

tests/ut/test_edge_groups.py

@@ -0,0 +1,23 @@
+from cterasdk.edge import groups
+from tests.ut import base_edge
+
+
+class TestEdgeGroups(base_edge.BaseEdgeTest):
+
+    def setUp(self):
+        super().setUp()
+        self._group_name = 'group'
+
+    def test_get_all_group(self):
+        get_response = 'Success'
+        self._init_filer(get_response=get_response)
+        ret = groups.Groups(self._filer).get()
+        self._filer.get.assert_called_once_with('/config/auth/groups')
+        self.assertEqual(ret, get_response)
+
+    def test_get_group(self):
+        get_response = 'Success'
+        self._init_filer(get_response=get_response)
+        ret = groups.Groups(self._filer).get(self._group_name)
+        self._filer.get.assert_called_once_with('/config/auth/groups/' + self._group_name)
+        self.assertEqual(ret, get_response)

tests/ut/test_edge_login.py

@@ -0,0 +1,49 @@
+from unittest import mock
+
+from cterasdk import exception
+from cterasdk.edge import login
+from tests.ut import base_edge
+
+
+class TestEdgeLogin(base_edge.BaseEdgeTest):
+
+    def setUp(self):
+        super().setUp()
+        self._username = 'admin'
+        self._password = 'password'
+
+    def test_login_success(self):
+        self._init_filer()
+        login.Login(self._filer).login(self._username, self._password)
+        self._filer.form_data.assert_called_once_with('/login', {'username': self._username, 'password': self._password})
+
+    def test_login_failure(self):
+        error_message = "Expected Failure"
+        expected_exception = exception.CTERAException(message=error_message)
+        self._filer.form_data = mock.MagicMock(side_effect=expected_exception)
+        with self.assertRaises(exception.CTERAException) as error:
+            login.Login(self._filer).login(self._username, self._password)
+        self.assertEqual(error_message, error.exception.message)
+
+    def test_logout_success_after_login_success(self):
+        self._init_filer()
+        login.Login(self._filer).login(self._username, self._password)
+        login.Login(self._filer).logout()
+        self._filer.form_data.assert_has_calls(
+            [
+                mock.call('/login', {'username': self._username, 'password': self._password}),
+                mock.call('/logout', {'foo': 'bar'})
+            ]
+        )
+
+    def test_logout_failure_after_login_success(self):
+        self._init_filer()
+        login.Login(self._filer).login(self._username, self._password)
+        self._filer.form_data.assert_called_once_with('/login', {'username': self._username, 'password': self._password})
+        error_message = "Expected Failure"
+        expected_exception = exception.CTERAException(message=error_message)
+        self._filer.form_data = mock.MagicMock(side_effect=expected_exception)
+        with self.assertRaises(exception.CTERAException) as error:
+            login.Login(self._filer).logout()
+        self._filer.form_data.assert_called_once_with('/logout', {'foo': 'bar'})
+        self.assertEqual(error_message, error.exception.message)