neil / rpms / udisks2

Forked from rpms/udisks2 a year ago
Clone
6903c7
commit fab797fcf5e4c8e09e4cde45647951acd764415e
6903c7
Author: Tomas Bzatek <tbzatek@redhat.com>
6903c7
Date:   Mon Oct 10 13:58:15 2022 +0200
6903c7
6903c7
    tests: Add bad auth test for iscsi
6903c7
    
6903c7
    This tests that the auth info is properly set for each login call,
6903c7
    overriding previously set auth info with no trace.
6903c7
6903c7
diff --git a/src/tests/dbus-tests/test_30_iscsi.py b/src/tests/dbus-tests/test_30_iscsi.py
6903c7
index 34bdfc4b..6ac8386b 100644
6903c7
--- a/src/tests/dbus-tests/test_30_iscsi.py
6903c7
+++ b/src/tests/dbus-tests/test_30_iscsi.py
6903c7
@@ -284,3 +284,61 @@ class UdisksISCSITest(udiskstestcase.UdisksTestCase):
6903c7
         # make sure the session object is no longer on dbus
6903c7
         objects = udisks.GetManagedObjects(dbus_interface='org.freedesktop.DBus.ObjectManager')
6903c7
         self.assertNotIn(session_path, objects.keys())
6903c7
+
6903c7
+    def test_login_noauth_badauth(self):
6903c7
+        """
6903c7
+        Test auth info override
6903c7
+        """
6903c7
+        manager = self.get_object('/Manager')
6903c7
+        nodes, _ = manager.DiscoverSendTargets(self.address, self.port, self.no_options,
6903c7
+                                               dbus_interface=self.iface_prefix + '.Manager.ISCSI.Initiator',
6903c7
+                                               timeout=self.iscsi_timeout)
6903c7
+
6903c7
+        node = next((node for node in nodes if node[0] == self.noauth_iqn), None)
6903c7
+        self.assertIsNotNone(node)
6903c7
+
6903c7
+        (iqn, tpg, host, port, iface) = node
6903c7
+        self.assertEqual(iqn, self.noauth_iqn)
6903c7
+        self.assertEqual(host, self.address)
6903c7
+        self.assertEqual(port, self.port)
6903c7
+
6903c7
+        self.addCleanup(self._force_lougout, self.noauth_iqn)
6903c7
+
6903c7
+        # first attempt - wrong password
6903c7
+        options = dbus.Dictionary(signature='sv')
6903c7
+        options['username'] = self.initiator
6903c7
+        msg = 'Login failed: initiator reported error'
6903c7
+        with six.assertRaisesRegex(self, dbus.exceptions.DBusException, msg):
6903c7
+            options['password'] = '12345'
6903c7
+            manager.Login(iqn, tpg, host, port, iface, options,
6903c7
+                          dbus_interface=self.iface_prefix + '.Manager.ISCSI.Initiator',
6903c7
+                          timeout=self.iscsi_timeout)
6903c7
+
6903c7
+        # second atttempt - no password
6903c7
+        manager.Login(iqn, tpg, host, port, iface, self.no_options,
6903c7
+                      dbus_interface=self.iface_prefix + '.Manager.ISCSI.Initiator',
6903c7
+                      timeout=self.iscsi_timeout)
6903c7
+
6903c7
+        devs = glob.glob('/dev/disk/by-path/*%s*' % iqn)
6903c7
+        self.assertEqual(len(devs), 1)
6903c7
+
6903c7
+        # check if the block device have 'Symlinks' property updated
6903c7
+        disk_name = os.path.realpath(devs[0]).split('/')[-1]
6903c7
+        disk_obj = self.get_object('/block_devices/' + disk_name)
6903c7
+        dbus_path = str(disk_obj.object_path)
6903c7
+        self.assertIsNotNone(disk_obj)
6903c7
+
6903c7
+        symlinks = self.get_property_raw(disk_obj, '.Block', 'Symlinks')
6903c7
+        self.assertIn(self.str_to_ay(devs[0]), symlinks)
6903c7
+
6903c7
+        manager.Logout(iqn, tpg, host, port, iface, self.no_options,
6903c7
+                       dbus_interface=self.iface_prefix + '.Manager.ISCSI.Initiator',
6903c7
+                       timeout=self.iscsi_timeout)
6903c7
+
6903c7
+        devs = glob.glob('/dev/disk/by-path/*%s*' % iqn)
6903c7
+        self.assertEqual(len(devs), 0)
6903c7
+
6903c7
+        # make sure the disk is no longer on dbus
6903c7
+        udisks = self.get_object('')
6903c7
+        objects = udisks.GetManagedObjects(dbus_interface='org.freedesktop.DBus.ObjectManager')
6903c7
+        self.assertNotIn(dbus_path, objects.keys())