1+ from __future__ import annotations
2+
3+ import pytest
4+
5+ from lib .commands import SSHCommandFailed
6+ from lib .common import Defer , vm_image , wait_for
7+ from lib .fistpoint import FistPoint
8+ from lib .host import Host
9+ from lib .sr import SR
10+ from lib .vdi import QCOW2_IMAGE_FORMAT , VDI
11+ from lib .vm import VM
12+ from tests .storage import (
13+ MAX_VDI_SIZE ,
14+ CoalesceOperation ,
15+ XVACompression ,
16+ coalesce_integrity ,
17+ full_vdi_write ,
18+ try_to_create_sr_with_missing_device ,
19+ vdi_export_import ,
20+ vdi_is_open ,
21+ xva_export_import ,
22+ )
23+
24+ # Requirements:
25+ # - one XCP-ng host >= 8.3 with an additional unused native 4KiB disk for the SR
26+
27+ class TestEXTSR4KCreateDestroy :
28+ """
29+ Tests that do not use fixtures that setup the SR or import VMs,
30+ because they precisely need to test SR creation and destruction,
31+ and VM import.
32+ """
33+
34+ def test_create_sr_with_missing_device (self , host : Host ) -> None :
35+ try_to_create_sr_with_missing_device ('ext' , 'EXT-4K-local-SR-test' , host )
36+
37+ def test_create_and_destroy_sr (self , host : Host ,
38+ unused_4k_disks : dict [Host , list [Host .BlockDeviceInfo ]]) -> None :
39+ # Create and destroy tested in the same test to leave the host as unchanged as possible
40+ sr_disk = unused_4k_disks [host ][0 ].name
41+ sr = host .sr_create ('ext' , "EXT-4K-local-SR-test" ,
42+ {'device' : '/dev/' + sr_disk ,
43+ 'preferred-image-formats' : QCOW2_IMAGE_FORMAT }, verify = True )
44+ # import a VM in order to detect vm import issues here rather than in the vm_on_xfs_fixture used in
45+ # the next tests, because errors in fixtures break teardown
46+ vm = host .import_vm (vm_image ('mini-linux-x86_64-bios' ), sr_uuid = sr .uuid )
47+ vm .destroy (verify = True )
48+ sr .destroy (verify = True )
49+
50+ @pytest .mark .usefixtures ("ext_sr_4k" )
51+ class TestEXTSR4K :
52+ @pytest .mark .quicktest
53+ def test_quicktest (self , ext_sr_4k : SR ) -> None :
54+ ext_sr_4k .run_quicktest ()
55+
56+ def test_vdi_is_not_open (self , vdi_on_ext_sr_4k : VDI ) -> None :
57+ assert not vdi_is_open (vdi_on_ext_sr_4k )
58+
59+ def test_vdi_image_format (self , vdi_on_ext_sr_4k : VDI ) -> None :
60+ fmt = vdi_on_ext_sr_4k .get_image_format ()
61+ # feature-detect: if the SM doesn't report image-format, skip this check
62+ if not fmt :
63+ pytest .skip ("SM does not report sm-config:image-format; skipping format check" )
64+ assert fmt == QCOW2_IMAGE_FORMAT
65+
66+ @pytest .mark .small_vm # run with a small VM to test the features
67+ @pytest .mark .big_vm # and ideally with a big VM to test it scales
68+ def test_start_and_shutdown_VM (self , vm_on_ext_sr_4k : VM ) -> None :
69+ vm = vm_on_ext_sr_4k
70+ vm .start ()
71+ vm .wait_for_os_booted ()
72+ vm .shutdown (verify = True )
73+
74+ @pytest .mark .small_vm
75+ @pytest .mark .big_vm
76+ def test_snapshot (self , vm_on_ext_sr_4k : VM ) -> None :
77+ vm = vm_on_ext_sr_4k
78+ vm .start ()
79+ try :
80+ vm .wait_for_os_booted ()
81+ vm .test_snapshot_on_running_vm ()
82+ finally :
83+ vm .shutdown (verify = True )
84+
85+ @pytest .mark .small_vm
86+ @pytest .mark .parametrize ("vdi_op" , ["snapshot" , "clone" ])
87+ def test_coalesce (self , storage_test_vm : VM , vdi_on_ext_sr_4k : VDI , vdi_op : CoalesceOperation ,
88+ defer : Defer ) -> None :
89+ coalesce_integrity (storage_test_vm , vdi_on_ext_sr_4k , vdi_op , defer )
90+
91+ @pytest .mark .small_vm
92+ @pytest .mark .parametrize ("compression" , ["none" , "gzip" , "zstd" ])
93+ def test_xva_export_import (self , vm_on_ext_sr_4k : VM , compression : XVACompression , temp_large_dir : str ,
94+ defer : Defer ) -> None :
95+ xva_export_import (vm_on_ext_sr_4k , compression , temp_large_dir , defer )
96+
97+ @pytest .mark .small_vm
98+ def test_vdi_export_import (self , storage_test_vm : VM , ext_sr_4k : SR , temp_large_dir : str ,
99+ defer : Defer ) -> None :
100+ vdi_export_import (storage_test_vm , ext_sr_4k , QCOW2_IMAGE_FORMAT , temp_large_dir , defer )
101+
102+ @pytest .mark .small_vm
103+ @pytest .mark .disk_throughput_intensive
104+ def test_full_vdi_write (self , storage_test_vm : VM , vdi_on_ext_sr_4k : VDI , defer : Defer ):
105+ full_vdi_write (storage_test_vm , vdi_on_ext_sr_4k , defer )
106+
107+ @pytest .mark .small_vm
108+ def test_invalid_vdi_size (self , ext_sr_4k : SR ):
109+ with pytest .raises (SSHCommandFailed ) as excinfo :
110+ ext_sr_4k .create_vdi (virtual_size = MAX_VDI_SIZE [QCOW2_IMAGE_FORMAT ] + 1 )
111+ assert 'VDI Invalid size' in excinfo .value .stdout
112+
113+ # *** tests with blktap activate failure (longer tests).
114+
115+ @pytest .mark .small_vm
116+ @pytest .mark .big_vm
117+ def test_blktap_activate_failure (self , vm_on_ext_sr_4k : VM ) -> None :
118+ vm = vm_on_ext_sr_4k
119+ with FistPoint (vm .host , "blktap_activate_inject_failure" ), pytest .raises (SSHCommandFailed ):
120+ vm .start ()
121+ vm .shutdown (force = True )
122+
123+ @pytest .mark .small_vm
124+ @pytest .mark .big_vm
125+ def test_resize (self , vm_on_ext_sr_4k : VM ) -> None :
126+ vm = vm_on_ext_sr_4k
127+ vdi = VDI (vm .vdi_uuids ()[0 ], host = vm .host )
128+ old_size = vdi .get_virtual_size ()
129+ new_size = old_size + (1 * 1024 * 1024 * 1024 ) # Adding a 1GiB to size
130+
131+ vdi .resize (new_size )
132+
133+ assert vdi .get_virtual_size () == new_size
134+
135+ @pytest .mark .reboot
136+ @pytest .mark .small_vm
137+ def test_reboot (self , host : Host , ext_sr_4k : SR , vm_on_ext_sr_4k : VM ) -> None :
138+ sr = ext_sr_4k
139+ vm = vm_on_ext_sr_4k
140+ host .reboot (verify = True )
141+ wait_for (sr .all_pbds_attached , "Wait for PBD attached" )
142+ # start the VM as a way to check that the underlying SR is operational
143+ vm .start ()
144+ vm .wait_for_os_booted ()
145+ vm .shutdown (verify = True )
146+
147+ # *** End of tests with reboots
0 commit comments