Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions tests/cold/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,3 +79,66 @@ def ca_crt_source_provider_inventory(
mtv_namespace=mtv_namespace,
provider=ca_crt_source_provider,
)


@pytest.fixture(scope="class")
def insecure_source_provider(
fixture_store: dict[str, Any],
session_uuid: str,
source_provider_data: dict[str, Any],
target_namespace: str,
ocp_admin_client: DynamicClient,
tmp_path_factory: pytest.TempPathFactory,
destination_ocp_secret: Secret, # pragma: allowlist secret
) -> Generator[BaseProvider, None, None]:
"""Source provider created with insecureSkipVerify=true (MTV-664).

Args:
fixture_store (dict[str, Any]): Session fixture store for resource tracking.
session_uuid (str): Unique session identifier.
source_provider_data (dict[str, Any]): Provider configuration from providers JSON.
target_namespace (str): Target namespace for provider resources.
ocp_admin_client (DynamicClient): OpenShift admin client.
tmp_path_factory (pytest.TempPathFactory): Temp directory factory for cert files.
destination_ocp_secret (Secret): Destination OCP cluster secret.

Yields:
BaseProvider: Source provider instance with insecureSkipVerify=true.
"""
with create_source_provider(
fixture_store=fixture_store,
session_uuid=session_uuid,
source_provider_data=source_provider_data,
namespace=target_namespace,
admin_client=ocp_admin_client,
tmp_dir=tmp_path_factory,
ocp_admin_client=ocp_admin_client,
destination_ocp_secret=destination_ocp_secret,
insecure=True,
) as _source_provider:
yield _source_provider

_source_provider.disconnect()


@pytest.fixture(scope="class")
def insecure_source_provider_inventory(
ocp_admin_client: DynamicClient,
mtv_namespace: str,
insecure_source_provider: BaseProvider,
) -> ForkliftInventory:
"""ForkliftInventory instance for the insecureSkipVerify provider (MTV-664).

Args:
ocp_admin_client (DynamicClient): OpenShift admin client.
mtv_namespace (str): MTV operator namespace.
insecure_source_provider (BaseProvider): Source provider with insecureSkipVerify=true.

Returns:
ForkliftInventory: Inventory instance for the insecure provider.
"""
return create_forklift_inventory(
client=ocp_admin_client,
mtv_namespace=mtv_namespace,
provider=insecure_source_provider,
)
284 changes: 284 additions & 0 deletions tests/cold/test_insecure_skip_verify_migration.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,284 @@
from __future__ import annotations

import base64
from typing import TYPE_CHECKING, Any

import pytest
from ocp_resources.network_map import NetworkMap
from ocp_resources.plan import Plan
from ocp_resources.secret import Secret
from ocp_resources.storage_map import StorageMap
from pytest_testconfig import config as py_config

from libs.base_provider import BaseProvider
from libs.forklift_inventory import ForkliftInventory
from utilities.mtv_migration import (
create_plan_resource,
get_network_migration_map,
get_storage_migration_map,
)
from utilities.utils import populate_vm_ids

if TYPE_CHECKING:
from kubernetes.dynamic import DynamicClient

from libs.providers.openshift import OCPProvider


@pytest.mark.vsphere
@pytest.mark.openstack
@pytest.mark.esxi
@pytest.mark.tier1
@pytest.mark.incremental
@pytest.mark.parametrize(
"class_plan_config",
[
pytest.param(
py_config["tests_params"]["test_insecure_skip_verify_cold_migration"],
)
],
indirect=True,
ids=["MTV-664"],
)
class TestInsecureSkipVerifyColdMigration:
"""Verify provider with insecureSkipVerify=true reaches plan readiness.

Test scenario:
1. A source provider is created with insecureSkipVerify=true in its Secret, bypassing
TLS certificate verification for the provider connection
2. The provider secret is verified to contain insecureSkipVerify set to "true"
3. StorageMap and NetworkMap resources are created using the insecure provider
4. A migration Plan is created using the insecure provider's StorageMap and NetworkMap

Expected result:
- Provider connects successfully to the source infrastructure with TLS verification disabled
- Plan reaches Ready status, confirming that the insecureSkipVerify flag is properly
propagated and honored by Forklift

This is a 4-step plan-readiness validation test (no migration executed).
"""

storage_map: StorageMap
network_map: NetworkMap
plan_resource: Plan

@pytest.mark.usefixtures("prepared_plan")
def test_verify_insecure_skip_verify(
self,
insecure_source_provider: BaseProvider,
ocp_admin_client: DynamicClient,
) -> None:
"""Verify the provider secret has insecureSkipVerify set to true."""
assert insecure_source_provider.ocp_resource is not None, "ocp_resource is not set"
secret_ref = insecure_source_provider.ocp_resource.instance.spec.secret
secret = Secret(
client=ocp_admin_client,
name=secret_ref.name,
namespace=secret_ref.namespace,
)
actual_value = base64.b64decode(secret.instance.data["insecureSkipVerify"]).decode("utf-8")
assert actual_value == "true", f"Expected insecureSkipVerify='true', got '{actual_value}'"

def test_create_storagemap(
self,
prepared_plan: dict[str, Any],
fixture_store: dict[str, Any],
ocp_admin_client: DynamicClient,
insecure_source_provider: BaseProvider,
destination_provider: OCPProvider,
insecure_source_provider_inventory: ForkliftInventory,
target_namespace: str,
) -> None:
"""Create StorageMap resource for migration."""
vms = [vm["name"] for vm in prepared_plan["virtual_machines"]]
self.__class__.storage_map = get_storage_migration_map(
fixture_store=fixture_store,
source_provider=insecure_source_provider,
destination_provider=destination_provider,
source_provider_inventory=insecure_source_provider_inventory,
ocp_admin_client=ocp_admin_client,
target_namespace=target_namespace,
vms=vms,
)
assert self.storage_map, "StorageMap creation failed"

def test_create_networkmap(
self,
prepared_plan: dict[str, Any],
fixture_store: dict[str, Any],
ocp_admin_client: DynamicClient,
insecure_source_provider: BaseProvider,
destination_provider: OCPProvider,
insecure_source_provider_inventory: ForkliftInventory,
target_namespace: str,
multus_network_name: dict[str, str],
) -> None:
"""Create NetworkMap resource for migration."""
vms = [vm["name"] for vm in prepared_plan["virtual_machines"]]
self.__class__.network_map = get_network_migration_map(
fixture_store=fixture_store,
source_provider=insecure_source_provider,
destination_provider=destination_provider,
source_provider_inventory=insecure_source_provider_inventory,
ocp_admin_client=ocp_admin_client,
target_namespace=target_namespace,
multus_network_name=multus_network_name,
vms=vms,
)
assert self.network_map, "NetworkMap creation failed"

def test_create_plan(
self,
prepared_plan: dict[str, Any],
fixture_store: dict[str, Any],
ocp_admin_client: DynamicClient,
insecure_source_provider: BaseProvider,
destination_provider: OCPProvider,
target_namespace: str,
insecure_source_provider_inventory: ForkliftInventory,
) -> None:
"""Create MTV Plan CR resource."""
populate_vm_ids(prepared_plan, insecure_source_provider_inventory)

self.__class__.plan_resource = create_plan_resource(
ocp_admin_client=ocp_admin_client,
fixture_store=fixture_store,
source_provider=insecure_source_provider,
destination_provider=destination_provider,
storage_map=self.storage_map,
network_map=self.network_map,
virtual_machines_list=prepared_plan["virtual_machines"],
target_namespace=target_namespace,
warm_migration=prepared_plan.get("warm_migration", False),
)
assert self.plan_resource, "Plan creation failed"


@pytest.mark.rhv
@pytest.mark.tier1
@pytest.mark.incremental
@pytest.mark.parametrize(
"class_plan_config",
[
pytest.param(
py_config["tests_params"]["test_insecure_skip_verify_cold_migration_rhv"],
)
],
indirect=True,
ids=["MTV-664-rhv"],
)
class TestInsecureSkipVerifyRhvColdMigration:
"""Verify provider with insecureSkipVerify=true reaches plan readiness (RHV).

RHV variant without skip_clone — RHV uses template names that require cloning
to produce VMs visible in the Forklift inventory.

Test scenario:
1. A source provider is created with insecureSkipVerify=true in its Secret, bypassing
TLS certificate verification for the provider connection
2. The provider secret is verified to contain insecureSkipVerify set to "true"
3. StorageMap and NetworkMap resources are created using the insecure provider
4. A migration Plan is created using the insecure provider's StorageMap and NetworkMap

Expected result:
- Provider connects successfully to the source infrastructure with TLS verification disabled
- Plan reaches Ready status, confirming that the insecureSkipVerify flag is properly
propagated and honored by Forklift

This is a 4-step plan-readiness validation test (no migration executed).
"""

storage_map: StorageMap
network_map: NetworkMap
plan_resource: Plan

@pytest.mark.usefixtures("prepared_plan")
def test_verify_insecure_skip_verify(
self,
insecure_source_provider: BaseProvider,
ocp_admin_client: DynamicClient,
) -> None:
Comment thread
coderabbitai[bot] marked this conversation as resolved.
"""Verify the provider secret has insecureSkipVerify set to true."""
assert insecure_source_provider.ocp_resource is not None, "ocp_resource is not set"
secret_ref = insecure_source_provider.ocp_resource.instance.spec.secret
secret = Secret(
client=ocp_admin_client,
name=secret_ref.name,
namespace=secret_ref.namespace,
)
actual_value = base64.b64decode(secret.instance.data["insecureSkipVerify"]).decode("utf-8")
assert actual_value == "true", f"Expected insecureSkipVerify='true', got '{actual_value}'"

def test_create_storagemap(
self,
prepared_plan: dict[str, Any],
fixture_store: dict[str, Any],
ocp_admin_client: DynamicClient,
insecure_source_provider: BaseProvider,
destination_provider: OCPProvider,
insecure_source_provider_inventory: ForkliftInventory,
target_namespace: str,
) -> None:
"""Create StorageMap resource for migration."""
vms = [vm["name"] for vm in prepared_plan["virtual_machines"]]
self.__class__.storage_map = get_storage_migration_map(
fixture_store=fixture_store,
source_provider=insecure_source_provider,
destination_provider=destination_provider,
source_provider_inventory=insecure_source_provider_inventory,
ocp_admin_client=ocp_admin_client,
target_namespace=target_namespace,
vms=vms,
)
assert self.storage_map, "StorageMap creation failed"

def test_create_networkmap(
self,
prepared_plan: dict[str, Any],
fixture_store: dict[str, Any],
ocp_admin_client: DynamicClient,
insecure_source_provider: BaseProvider,
destination_provider: OCPProvider,
insecure_source_provider_inventory: ForkliftInventory,
target_namespace: str,
multus_network_name: dict[str, str],
) -> None:
"""Create NetworkMap resource for migration."""
vms = [vm["name"] for vm in prepared_plan["virtual_machines"]]
self.__class__.network_map = get_network_migration_map(
fixture_store=fixture_store,
source_provider=insecure_source_provider,
destination_provider=destination_provider,
source_provider_inventory=insecure_source_provider_inventory,
ocp_admin_client=ocp_admin_client,
target_namespace=target_namespace,
multus_network_name=multus_network_name,
vms=vms,
)
assert self.network_map, "NetworkMap creation failed"

def test_create_plan(
self,
prepared_plan: dict[str, Any],
fixture_store: dict[str, Any],
ocp_admin_client: DynamicClient,
insecure_source_provider: BaseProvider,
destination_provider: OCPProvider,
target_namespace: str,
insecure_source_provider_inventory: ForkliftInventory,
) -> None:
"""Create MTV Plan CR resource."""
populate_vm_ids(prepared_plan, insecure_source_provider_inventory)

Comment thread
qodo-code-review[bot] marked this conversation as resolved.
self.__class__.plan_resource = create_plan_resource(
ocp_admin_client=ocp_admin_client,
fixture_store=fixture_store,
source_provider=insecure_source_provider,
destination_provider=destination_provider,
storage_map=self.storage_map,
network_map=self.network_map,
virtual_machines_list=prepared_plan["virtual_machines"],
target_namespace=target_namespace,
warm_migration=prepared_plan.get("warm_migration", False),
)
assert self.plan_resource, "Plan creation failed"
13 changes: 13 additions & 0 deletions tests/tests_config/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -782,6 +782,19 @@
],
"warm_migration": False,
},
"test_insecure_skip_verify_cold_migration": {
"virtual_machines": [
{"name": "mtv-tests-rhel8", "guest_agent": True},
],
"warm_migration": False,
"skip_clone": True,
Comment thread
qodo-code-review[bot] marked this conversation as resolved.
},
"test_insecure_skip_verify_cold_migration_rhv": {
"virtual_machines": [
{"name": "mtv-tests-rhel8", "guest_agent": True},
],
"warm_migration": False,
},
"test_upgrade_cold_migration": {
"virtual_machines": [
{"name": "mtv-tests-rhel8", "guest_agent": True},
Expand Down