lvmdbustest.py: Add unit test for external VG create

Ensure a VG created outside of the dbus service is immediately found by
service user.

diff --git a/test/dbus/lvmdbustest.py b/test/dbus/lvmdbustest.py
index 71b62e1514999b6d90f67b17ee4b40242586c569..4788eb3715aa0ad4dd9d2dbdc0e779e6ce8ec88a 100755 (executable)
--- a/test/dbus/lvmdbustest.py
+++ b/test/dbus/lvmdbustest.py
@@ -17,6 +17,7 @@ import unittest
 import pyudev
 from testlib import *
 import testlib
+from subprocess import Popen, PIPE
 
 g_tmo = 0
 
@@ -35,6 +36,9 @@ pv_device_list = os.getenv('LVM_DBUSD_PV_DEVICE_LIST', None)
 # Other == Test just lvm shell mode
 test_shell = os.getenv('LVM_DBUSD_TEST_MODE', 1)
 
+# LVM binary to use
+LVM_EXECUTABLE = os.getenv('LVM_BINARY', '/usr/sbin/lvm')
+
 # Empty options dictionary (EOD)
 EOD = dbus.Dictionary({}, signature=dbus.Signature('sv'))
 # Base interfaces on LV objects
@@ -111,6 +115,26 @@ def set_execution(lvmshell, test_result):
        return rc
 
 
+def call_lvm(command):
+       """
+       Call lvm executable and return a tuple of exitcode, stdout, stderr
+       :param command:     Command to execute
+       :param debug:       Dump debug to stdout
+       """
+
+       # Prepend the full lvm executable so that we can run different versions
+       # in different locations on the same box
+       command.insert(0, LVM_EXECUTABLE)
+
+       process = Popen(command, stdout=PIPE, stderr=PIPE, close_fds=True,
+                                       env=os.environ)
+       out = process.communicate()
+
+       stdout_text = bytes(out[0]).decode("utf-8")
+       stderr_text = bytes(out[1]).decode("utf-8")
+       return process.returncode, stdout_text, stderr_text
+
+
 # noinspection PyUnresolvedReferences
 class TestDbusService(unittest.TestCase):
        def setUp(self):
@@ -1702,6 +1726,29 @@ class TestDbusService(unittest.TestCase):
                        tag in vg_proxy.Vg.Tags,
                        "%s not in %s" % (tag, str(vg_proxy.Vg.Tags)))
 
+       def _verify_existence(self, cmd, operation, resource_name):
+               ec, stdout, stderr = call_lvm(cmd)
+               if ec == 0:
+                       path = self._lookup(resource_name)
+                       self.assertTrue(path != '/')
+               else:
+                       self.assertTrue(ec == 0, "%s exit code = %d" % (operation, ec))
+                       std_err_print(
+                               "%s failed with stdout= %s, stderr= %s" %
+                               (operation, stdout, stderr))
+
+       def test_external_vg_create(self):
+               # We need to ensure that if a user creates something outside of lvm
+               # dbus service that things are sequenced correctly so that if a dbus
+               # user calls into the service they will find the same information.
+               vg_name = vg_n()
+
+               # Get all the PV device paths
+               pv_paths = [p.Pv.Name for p in self.objs[PV_INT]]
+
+               cmd = ['vgcreate', vg_name]
+               cmd.extend(pv_paths)
+               self._verify_existence(cmd, cmd[0], vg_name)
 
 class AggregateResults(object):