test_ext_sr.py

from __future__ import annotations

import pytest

import logging

from lib.commands import SSHCommandFailed
from lib.common import Defer, vm_image, wait_for
from lib.fistpoint import FistPoint
from lib.host import Host
from lib.sr import SR
from lib.vdi import VDI
from lib.vm import VM
from tests.storage import (
    MAX_VDI_SIZE,
    CBTTest,
    CoalesceOperation,
    ImageFormat,
    XVACompression,
    assert_cbt_log_does_not_exist_file_sr,
    assert_cbt_log_exists_file_sr,
    coalesce_integrity,
    full_vdi_write,
    try_to_create_sr_with_missing_device,
    vdi_export_import,
    vdi_is_open,
    xva_export_import,
)

# Requirements:
# - one XCP-ng host with an additional unused disk for the SR

@pytest.mark.usefixtures("ext_sr")
class TestEXTSR:
    @pytest.mark.quicktest
    def test_quicktest(self, ext_sr: SR) -> None:
        ext_sr.run_quicktest()

    def test_vdi_is_not_open(self, vdi_on_ext_sr: VDI) -> None:
        assert not vdi_is_open(vdi_on_ext_sr)

    def test_vdi_image_format(self, vdi_on_ext_sr: VDI, image_format: ImageFormat) -> None:
        fmt = vdi_on_ext_sr.get_image_format()
        # feature-detect: if the SM doesn't report image-format, skip this check
        if not fmt:
            pytest.skip("SM does not report sm-config:image-format; skipping format check")
        assert fmt == image_format

    @pytest.mark.small_vm # run with a small VM to test the features
    @pytest.mark.big_vm # and ideally with a big VM to test it scales
    def test_start_and_shutdown_VM(self, vm_on_ext_sr: VM) -> None:
        vm = vm_on_ext_sr
        vm.start()
        vm.wait_for_os_booted()
        vm.shutdown(verify=True)

    @pytest.mark.small_vm
    @pytest.mark.big_vm
    def test_snapshot(self, vm_on_ext_sr: VM) -> None:
        vm = vm_on_ext_sr
        vm.start()
        try:
            vm.wait_for_os_booted()
            vm.test_snapshot_on_running_vm()
        finally:
            vm.shutdown(verify=True)

    @pytest.mark.small_vm
    @pytest.mark.parametrize("vdi_op", ["snapshot", "clone"])
    def test_coalesce(self, storage_test_vm: VM, vdi_on_ext_sr: VDI, vdi_op: CoalesceOperation, defer: Defer) -> None:
        coalesce_integrity(storage_test_vm, vdi_on_ext_sr, vdi_op, defer)

    @pytest.mark.small_vm
    @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
    def test_xva_export_import(self, vm_on_ext_sr: VM, compression: XVACompression, temp_large_dir: str, defer: Defer) \
            -> None:
        xva_export_import(vm_on_ext_sr, compression, temp_large_dir, defer)

    @pytest.mark.small_vm
    def test_vdi_export_import(self, storage_test_vm: VM, ext_sr: SR, image_format: ImageFormat, temp_large_dir: str,
                               defer: Defer) -> None:
        vdi_export_import(storage_test_vm, ext_sr, image_format, temp_large_dir, defer)

    @pytest.mark.small_vm
    @pytest.mark.disk_throughput_intensive
    def test_full_vdi_write(self, storage_test_vm: VM, vdi_on_ext_sr: VDI, defer: Defer):
        full_vdi_write(storage_test_vm, vdi_on_ext_sr, defer)

    @pytest.mark.small_vm
    def test_invalid_vdi_size(self, ext_sr: SR, image_format: ImageFormat):
        with pytest.raises(SSHCommandFailed) as excinfo:
            ext_sr.create_vdi(virtual_size=MAX_VDI_SIZE[image_format] + 1)
        assert 'VDI Invalid size' in excinfo.value.stdout

    # *** tests with reboots (longer tests).

    @pytest.mark.small_vm
    @pytest.mark.big_vm
    def test_blktap_activate_failure(self, vm_on_ext_sr: VM) -> None:
        from lib.fistpoint import FistPoint
        vm = vm_on_ext_sr
        with FistPoint(vm.host, "blktap_activate_inject_failure"), pytest.raises(SSHCommandFailed):
            vm.start()
            vm.shutdown(force=True)

    @pytest.mark.small_vm
    @pytest.mark.big_vm
    def test_resize(self, vm_on_ext_sr: VM) -> None:
        vm = vm_on_ext_sr
        vdi = VDI(vm.vdi_uuids()[0], host=vm.host)
        old_size = vdi.get_virtual_size()
        new_size = old_size + (1 * 1024 * 1024 * 1024) # Adding a 1GiB to size

        vdi.resize(new_size)

        assert vdi.get_virtual_size() == new_size

    @pytest.mark.small_vm
    @pytest.mark.big_vm
    def test_failing_resize(self, host: Host, ext_sr: SR, vm_on_ext_sr: VM, exit_on_fistpoint: None) -> None:
        vm = vm_on_ext_sr
        vdi = VDI(vm.vdi_uuids()[0], host=vm.host)
        old_size = vdi.get_virtual_size()
        new_size = old_size + (1 * 1024 * 1024 * 1024) # Adding a 1GiB to size

        with FistPoint(vm.host, "LVHDRT_inflate_after_setSize"):
            try:
                vdi.resize(new_size)
            except SSHCommandFailed:
                logging.info(f"Launching SR scan for {ext_sr} after failure")
                host.xe("sr-scan", {"uuid": ext_sr.uuid})

        assert vdi.get_virtual_size() == new_size

    @pytest.mark.reboot
    @pytest.mark.small_vm
    def test_reboot(self, host: Host, ext_sr: SR, vm_on_ext_sr: VM) -> None:
        sr = ext_sr
        vm = vm_on_ext_sr
        host.reboot(verify=True)
        wait_for(sr.all_pbds_attached, "Wait for PBD attached")
        # start the VM as a way to check that the underlying SR is operational
        vm.start()
        vm.wait_for_os_booted()
        vm.shutdown(verify=True)

    # *** End of tests with reboots


class TestEXTCBT(CBTTest):
    """Test CBT functionality on EXT SR"""

    @staticmethod
    def assert_cbt_log_exists(host: Host, sr: SR, vdi: VDI) -> None:
        assert_cbt_log_exists_file_sr(host, sr, vdi)

    @staticmethod
    def assert_cbt_log_does_not_exist(host: Host, sr: SR, vdi: VDI) -> None:
        assert_cbt_log_does_not_exist_file_sr(host, sr, vdi)

    def test_enable_disable_cbt(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_enable_disable_cbt(host, ext_sr, vdi_on_ext_sr)

    def test_cbt_log_creation(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_cbt_log_creation(host, ext_sr, vdi_on_ext_sr)

    def test_snapshot_with_cbt(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_snapshot_with_cbt(host, ext_sr, vdi_on_ext_sr)

    @pytest.mark.small_vm
    def test_changed_blocks_tracking(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI, vm_on_ext_sr: VM) -> None:
        self._test_changed_blocks_tracking(host, ext_sr, vdi_on_ext_sr, vm_on_ext_sr)

    @pytest.mark.small_vm
    def test_cbt_after_coalesce(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI, vm_on_ext_sr: VM) -> None:
        self._test_cbt_after_coalesce(host, ext_sr, vdi_on_ext_sr, vm_on_ext_sr)

    @pytest.mark.small_vm
    def test_incremental_snap_scenario(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI, vm_on_ext_sr: VM) -> None:
        self._test_incremental_snap_scenario(host, ext_sr, vdi_on_ext_sr, vm_on_ext_sr)

    def test_disable_cbt_removes_log(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_disable_cbt_removes_log(host, ext_sr, vdi_on_ext_sr)

    def test_destroy_vdi_removes_cbt_log(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_destroy_vdi_removes_cbt_log(host, ext_sr, vdi_on_ext_sr)

    def test_cbt_persist_after_sr_reboot(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_cbt_persist_after_sr_reboot(host, ext_sr, vdi_on_ext_sr)

    def test_cbt_on_snapshot_chain(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_cbt_on_snapshot_chain(host, ext_sr, vdi_on_ext_sr)

    def test_cbt_parent_disable_does_not_affect_snapshot(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_cbt_parent_disable_does_not_affect_snapshot(host, ext_sr, vdi_on_ext_sr)

    @pytest.mark.small_vm
    def test_cbt_bitmap_non_zero_after_write(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI,
                                             vm_on_ext_sr: VM) -> None:
        self._test_cbt_bitmap_non_zero_after_write(host, ext_sr, vdi_on_ext_sr, vm_on_ext_sr)

    def test_cbt_data_destroy(self, host: Host, ext_sr: SR, vdi_on_ext_sr: VDI) -> None:
        self._test_cbt_data_destroy(host, ext_sr, vdi_on_ext_sr)