mirror of
https://github.com/libguestfs/libguestfs.git
synced 2026-03-21 22:53:37 +00:00
python: tests: Use the 'unittest' module to run the test suite.
This commit is contained in:
@@ -15,4 +15,11 @@
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
|
||||
import guestfs
|
||||
import unittest
|
||||
|
||||
class Test010Load (unittest.TestCase):
|
||||
def test_import (self):
|
||||
import guestfs
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -15,11 +15,25 @@
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
g.add_drive ("/dev/null")
|
||||
g.add_drive ("/dev/null", readonly = True)
|
||||
g.add_drive ("/dev/null", iface = "virtio", format = "raw")
|
||||
g.close ()
|
||||
class Test070OptArgs (unittest.TestCase):
|
||||
def setUp (self):
|
||||
self.g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
def test_no_optargs (self):
|
||||
self.g.add_drive ("/dev/null")
|
||||
|
||||
def test_one_optarg (self):
|
||||
self.g.add_drive ("/dev/null", readonly = True)
|
||||
|
||||
def test_two_optargs (self):
|
||||
self.g.add_drive ("/dev/null", iface = "virtio", format = "raw")
|
||||
|
||||
def tearDown (self):
|
||||
self.g.close ()
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -15,18 +15,23 @@
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
g.add_drive_scratch (500 * 1024 * 1024)
|
||||
g.launch ()
|
||||
class Test100Launch (unittest.TestCase):
|
||||
def test_launch (self):
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
g.add_drive_scratch (500 * 1024 * 1024)
|
||||
g.launch ()
|
||||
|
||||
g.pvcreate ("/dev/sda")
|
||||
g.vgcreate ("VG", ["/dev/sda"])
|
||||
g.lvcreate ("LV1", "VG", 200)
|
||||
g.lvcreate ("LV2", "VG", 200)
|
||||
if (g.lvs () != ["/dev/VG/LV1", "/dev/VG/LV2"]):
|
||||
raise "Error: g.lvs() returned incorrect result"
|
||||
g.shutdown ()
|
||||
g.close ()
|
||||
g.pvcreate ("/dev/sda")
|
||||
g.vgcreate ("VG", ["/dev/sda"])
|
||||
g.lvcreate ("LV1", "VG", 200)
|
||||
g.lvcreate ("LV2", "VG", 200)
|
||||
self.assertEqual (g.lvs (), ["/dev/VG/LV1", "/dev/VG/LV2"])
|
||||
g.shutdown ()
|
||||
g.close ()
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -15,23 +15,27 @@
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
close_invoked = 0
|
||||
|
||||
def close_callback (ev, eh, buf, array):
|
||||
global close_invoked
|
||||
close_invoked += 1
|
||||
|
||||
# Register a callback for the close event.
|
||||
g.set_event_callback (close_callback, guestfs.EVENT_CLOSE)
|
||||
class Test410CloseEvent (unittest.TestCase):
|
||||
def test_close_event (self):
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
# Close the handle. The close callback should be invoked.
|
||||
if close_invoked != 0:
|
||||
raise "Error: close_invoked should be 0"
|
||||
g.close ()
|
||||
if close_invoked != 1:
|
||||
raise "Error: close_invoked should be 1"
|
||||
# Register a callback for the close event.
|
||||
g.set_event_callback (close_callback, guestfs.EVENT_CLOSE)
|
||||
|
||||
# Close the handle. The close callback should be invoked.
|
||||
self.assertEqual (close_invoked, 0)
|
||||
g.close ()
|
||||
self.assertEqual (close_invoked, 1)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -15,11 +15,10 @@
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
log_invoked = 0
|
||||
|
||||
def log_callback (ev,eh,buf,array):
|
||||
@@ -33,20 +32,26 @@ def log_callback (ev,eh,buf,array):
|
||||
print ("python event logged: event=%s eh=%d buf='%s' array=%s" %
|
||||
(guestfs.event_to_string (ev), eh, buf, array))
|
||||
|
||||
# Register an event callback for all log messages.
|
||||
events = guestfs.EVENT_APPLIANCE | guestfs.EVENT_LIBRARY \
|
||||
| guestfs.EVENT_WARNING | guestfs.EVENT_TRACE
|
||||
g.set_event_callback (log_callback, events)
|
||||
class Test420LogMessages (unittest.TestCase):
|
||||
def test_log_messages (self):
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
# Now make sure we see some messages.
|
||||
g.set_trace (1)
|
||||
g.set_verbose (1)
|
||||
# Register an event callback for all log messages.
|
||||
events = guestfs.EVENT_APPLIANCE | guestfs.EVENT_LIBRARY \
|
||||
| guestfs.EVENT_WARNING | guestfs.EVENT_TRACE
|
||||
g.set_event_callback (log_callback, events)
|
||||
|
||||
# Do some stuff.
|
||||
g.add_drive_ro ("/dev/null")
|
||||
g.set_autosync (1)
|
||||
# Now make sure we see some messages.
|
||||
g.set_trace (1)
|
||||
g.set_verbose (1)
|
||||
|
||||
g.close ()
|
||||
# Do some stuff.
|
||||
g.add_drive_ro ("/dev/null")
|
||||
g.set_autosync (1)
|
||||
|
||||
if log_invoked == 0:
|
||||
raise "Error: log_invoked should be > 0"
|
||||
g.close ()
|
||||
|
||||
self.assertFalse (log_invoked == 0)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -17,43 +17,45 @@
|
||||
|
||||
# Test implicit vs explicit closes of the handle (RHBZ#717786).
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
g.close () # explicit close
|
||||
del g # implicit close - should be no error/warning
|
||||
|
||||
# Expect an exception if we call a method on a closed handle.
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
g.close ()
|
||||
try:
|
||||
g.set_memsize (512)
|
||||
raise Exception("expected an exception from previous statement")
|
||||
except guestfs.ClosedHandle:
|
||||
pass
|
||||
del g
|
||||
|
||||
# Verify that the handle is really being closed by g.close, by setting
|
||||
# up a close event and testing that it happened.
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
close_invoked = 0
|
||||
|
||||
def close_callback (ev, eh, buf, array):
|
||||
global close_invoked
|
||||
close_invoked += 1
|
||||
|
||||
g.set_event_callback (close_callback, guestfs.EVENT_CLOSE)
|
||||
class Test800ExplicitClose (unittest.TestCase):
|
||||
def test_explicit_close (self):
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
if close_invoked != 0:
|
||||
raise Exception("close_invoked should be 0")
|
||||
g.close () # explicit close
|
||||
del g # implicit close - should be no error/warning
|
||||
|
||||
g.close ()
|
||||
if close_invoked != 1:
|
||||
raise Exception("close_invoked should be 1")
|
||||
# Expect an exception if we call a method on a closed handle.
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
g.close ()
|
||||
try:
|
||||
g.set_memsize (512)
|
||||
raise Exception("expected an exception from previous statement")
|
||||
except guestfs.ClosedHandle:
|
||||
pass
|
||||
del g
|
||||
|
||||
del g
|
||||
if close_invoked != 1:
|
||||
raise Exception("close_invoked should be 1")
|
||||
# Verify that the handle is really being closed by g.close, by
|
||||
# setting up a close event and testing that it happened.
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
g.set_event_callback (close_callback, guestfs.EVENT_CLOSE)
|
||||
|
||||
self.assertEqual (close_invoked, 0)
|
||||
g.close ()
|
||||
self.assertEqual (close_invoked, 1)
|
||||
|
||||
del g
|
||||
self.assertEqual (close_invoked, 1)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -15,21 +15,29 @@
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
class Test810RHBZ811650 (unittest.TestCase):
|
||||
def test_rhbz811650 (self):
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
g.disk_create ("rhbz811650.img", "raw", 500 * 1024 * 1024)
|
||||
g.disk_create ("rhbz811650.img", "raw", 500 * 1024 * 1024)
|
||||
|
||||
# Deliberate error: the disk format is supposed to be raw.
|
||||
g.add_drive ("rhbz811650.img", format="qcow2");
|
||||
# Deliberate error: the disk format is supposed to be raw.
|
||||
g.add_drive ("rhbz811650.img", format="qcow2");
|
||||
|
||||
# Because error() wasn't being called, guestfs_last_error would return
|
||||
# NULL, causing a segfault in the Python bindings (RHBZ#811650).
|
||||
try:
|
||||
g.launch ()
|
||||
except:
|
||||
pass
|
||||
# Because error() wasn't being called, guestfs_last_error
|
||||
# would return NULL, causing a segfault in the Python bindings
|
||||
# (RHBZ#811650).
|
||||
try:
|
||||
g.launch ()
|
||||
except:
|
||||
pass
|
||||
|
||||
os.unlink ("rhbz811650.img")
|
||||
def tearDown (self):
|
||||
os.unlink ("rhbz811650.img")
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
# https://bugzilla.redhat.com/1075164#c7
|
||||
|
||||
from subprocess import check_output
|
||||
import unittest
|
||||
import random
|
||||
import string
|
||||
import re
|
||||
@@ -54,16 +55,19 @@ if not "c_pointer" in dir (conn):
|
||||
print ("skipping test: libvirt-python doesn't support c_pointer()")
|
||||
exit (77)
|
||||
|
||||
# Create a test disk.
|
||||
filename = os.getcwd () + "/820-rhbz912499.img"
|
||||
guestfs.GuestFS().disk_create (filename, "raw", 1024*1024*1024)
|
||||
class Test820RHBZ912499 (unittest.TestCase):
|
||||
def setUp (self):
|
||||
# Create a test disk.
|
||||
self.filename = os.getcwd () + "/820-rhbz912499.img"
|
||||
guestfs.GuestFS().disk_create (self.filename, "raw", 1024*1024*1024)
|
||||
|
||||
# Create a new domain. This won't work, it will just hang when
|
||||
# booted. But that's sufficient for the test.
|
||||
domname = ''.join (random.choice (string.ascii_uppercase) for _ in range (8))
|
||||
domname = "tmp-" + domname
|
||||
# Create a new domain. This won't work, it will just hang when
|
||||
# booted. But that's sufficient for the test.
|
||||
self.domname = ''.join (random.choice (string.ascii_uppercase)
|
||||
for _ in range (8))
|
||||
self.domname = "tmp-" + self.domname
|
||||
|
||||
xml = """
|
||||
self.xml = """
|
||||
<domain type='qemu'>
|
||||
<name>%s</name>
|
||||
<memory>1048576</memory>
|
||||
@@ -80,32 +84,35 @@ xml = """
|
||||
</disk>
|
||||
</devices>
|
||||
</domain>
|
||||
""" % (domname, filename)
|
||||
""" % (self.domname, self.filename)
|
||||
|
||||
dom = conn.createXML (xml, libvirt.VIR_DOMAIN_START_AUTODESTROY)
|
||||
if dom == None:
|
||||
raise "could not create temporary domain (%s)" % domname
|
||||
def test_rhbz912499 (self):
|
||||
dom = conn.createXML (self.xml,
|
||||
libvirt.VIR_DOMAIN_START_AUTODESTROY)
|
||||
self.assertFalse (dom == None)
|
||||
|
||||
print ("temporary domain %s is running" % domname)
|
||||
print ("temporary domain %s is running" % self.domname)
|
||||
|
||||
# Libvirt should have labelled the disk.
|
||||
print ("before starting libguestfs")
|
||||
before = check_output (["ls", "-Z", filename])
|
||||
print ("disk label = %s" % before)
|
||||
# Libvirt should have labelled the disk.
|
||||
print ("before starting libguestfs")
|
||||
before = check_output (["ls", "-Z", self.filename])
|
||||
print ("disk label = %s" % before)
|
||||
|
||||
# Now see if we can open the domain with libguestfs without
|
||||
# disturbing the label.
|
||||
g = guestfs.GuestFS ()
|
||||
r = g.add_libvirt_dom (dom, readonly = 1)
|
||||
if r != 1:
|
||||
raise "unexpected return value from add_libvirt_dom (%d)" % r
|
||||
g.launch ()
|
||||
# Now see if we can open the domain with libguestfs without
|
||||
# disturbing the label.
|
||||
g = guestfs.GuestFS ()
|
||||
r = g.add_libvirt_dom (dom, readonly = 1)
|
||||
self.assertEqual (r, 1)
|
||||
g.launch ()
|
||||
|
||||
print ("after starting libguestfs")
|
||||
after = check_output (["ls", "-Z", filename])
|
||||
print ("disk label = %s" % after)
|
||||
print ("after starting libguestfs")
|
||||
after = check_output (["ls", "-Z", self.filename])
|
||||
print ("disk label = %s" % after)
|
||||
|
||||
if before != after:
|
||||
raise "disk label was changed unexpectedly"
|
||||
self.assertEqual (before, after)
|
||||
|
||||
os.unlink (filename)
|
||||
def tearDown (self):
|
||||
os.unlink (self.filename)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -18,17 +18,30 @@
|
||||
# Test python-specific python_return_dict parameter.
|
||||
|
||||
from types import *
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=False)
|
||||
class Test900PythonDict (unittest.TestCase):
|
||||
def test_python_no_dict (self):
|
||||
g = guestfs.GuestFS (python_return_dict=False)
|
||||
|
||||
r = g.internal_test_rhashtable ("5")
|
||||
if type(r) != list or r != [ ("0","0"), ("1","1"), ("2","2"), ("3","3"), ("4","4") ]:
|
||||
raise Exception ("python_return_dict=False: internal_test_rhashtable returned %s" % r)
|
||||
r = g.internal_test_rhashtable ("5")
|
||||
self.assertTrue (isinstance (r, list))
|
||||
self.assertEqual (r, [ ("0","0"), ("1","1"), ("2","2"),
|
||||
("3","3"), ("4","4") ])
|
||||
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
def test_python_dict (self):
|
||||
g = guestfs.GuestFS (python_return_dict=True)
|
||||
|
||||
r = g.internal_test_rhashtable ("5")
|
||||
if type(r) != dict or sorted (r.keys()) != ["0","1","2","3","4"] or r["0"] != "0" or r["1"] != "1" or r["2"] != "2" or r["3"] != "3" or r["4"] != "4":
|
||||
raise Exception ("python_return_dict=True: internal_test_rhashtable returned %s" % r)
|
||||
r = g.internal_test_rhashtable ("5")
|
||||
self.assertTrue (isinstance (r, dict))
|
||||
self.assertEqual (sorted (r.keys()), ["0","1","2","3","4"])
|
||||
self.assertEqual (r["0"], "0")
|
||||
self.assertEqual (r["1"], "1")
|
||||
self.assertEqual (r["2"], "2")
|
||||
self.assertEqual (r["3"], "3")
|
||||
self.assertEqual (r["4"], "4")
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
# working by testing it. See:
|
||||
# https://bugzilla.redhat.com/show_bug.cgi?id=1075164
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import guestfs
|
||||
|
||||
@@ -38,11 +39,14 @@ if not "c_pointer" in dir (conn):
|
||||
print ("skipping test: libvirt-python doesn't support c_pointer()")
|
||||
exit (77)
|
||||
|
||||
dom = conn.lookupByName ("blank-disk")
|
||||
class Test910Libvirt (unittest.TestCase):
|
||||
def test_libvirt (self):
|
||||
dom = conn.lookupByName ("blank-disk")
|
||||
|
||||
g = guestfs.GuestFS ()
|
||||
g = guestfs.GuestFS ()
|
||||
|
||||
r = g.add_libvirt_dom (dom, readonly=1)
|
||||
r = g.add_libvirt_dom (dom, readonly=1)
|
||||
self.assertEqual (r, 1)
|
||||
|
||||
if r != 1:
|
||||
raise "unexpected return value from add_libvirt_dom (%d)" % r
|
||||
if __name__ == '__main__':
|
||||
unittest.main ()
|
||||
|
||||
Reference in New Issue
Block a user