570 lines
21 KiB
Python
570 lines
21 KiB
Python
"""Unit tests for the HarvesterPlugin provider plugin."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from iac_reverse.models import (
|
|
CpuArchitecture,
|
|
PlatformCategory,
|
|
ProviderType,
|
|
ScanProgress,
|
|
)
|
|
from iac_reverse.scanner.harvester_plugin import HarvesterPlugin
|
|
|
|
|
|
# Patch targets for kubernetes client classes
|
|
PATCH_NEW_CLIENT = "iac_reverse.scanner.harvester_plugin.config.new_client_from_config"
|
|
PATCH_CUSTOM_API = "iac_reverse.scanner.harvester_plugin.client.CustomObjectsApi"
|
|
PATCH_CORE_API = "iac_reverse.scanner.harvester_plugin.client.CoreV1Api"
|
|
|
|
|
|
class TestHarvesterPluginAuthentication:
|
|
"""Tests for HarvesterPlugin.authenticate()."""
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_authenticate_with_kubeconfig_path(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""Authenticate loads kubeconfig from the provided path."""
|
|
mock_api_client = MagicMock()
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
mock_new_client.assert_called_once_with(
|
|
config_file="/path/to/kubeconfig",
|
|
context=None,
|
|
)
|
|
assert plugin._api_client is mock_api_client
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_authenticate_with_context(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""Authenticate uses the optional context parameter."""
|
|
mock_api_client = MagicMock()
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({
|
|
"kubeconfig_path": "/path/to/kubeconfig",
|
|
"context": "harvester-cluster",
|
|
})
|
|
|
|
mock_new_client.assert_called_once_with(
|
|
config_file="/path/to/kubeconfig",
|
|
context="harvester-cluster",
|
|
)
|
|
|
|
def test_authenticate_missing_kubeconfig_path(self):
|
|
"""Authenticate raises AuthenticationError when kubeconfig_path is missing."""
|
|
from iac_reverse.scanner import AuthenticationError
|
|
|
|
plugin = HarvesterPlugin()
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({})
|
|
|
|
assert "kubeconfig_path" in str(exc_info.value)
|
|
assert "harvester" in str(exc_info.value)
|
|
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_authenticate_invalid_kubeconfig(self, mock_new_client):
|
|
"""Authenticate raises AuthenticationError when kubeconfig is invalid."""
|
|
from iac_reverse.scanner import AuthenticationError
|
|
|
|
mock_new_client.side_effect = Exception("Invalid kubeconfig format")
|
|
|
|
plugin = HarvesterPlugin()
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
plugin.authenticate({"kubeconfig_path": "/bad/path"})
|
|
|
|
assert "Failed to load kubeconfig" in str(exc_info.value)
|
|
|
|
|
|
class TestHarvesterPluginMetadata:
|
|
"""Tests for HarvesterPlugin metadata methods."""
|
|
|
|
def test_get_platform_category(self):
|
|
"""get_platform_category returns HCI."""
|
|
plugin = HarvesterPlugin()
|
|
assert plugin.get_platform_category() == PlatformCategory.HCI
|
|
|
|
def test_list_supported_resource_types(self):
|
|
"""list_supported_resource_types returns all Harvester resource types."""
|
|
plugin = HarvesterPlugin()
|
|
types = plugin.list_supported_resource_types()
|
|
|
|
assert types == [
|
|
"harvester_virtualmachine",
|
|
"harvester_volume",
|
|
"harvester_image",
|
|
"harvester_network",
|
|
]
|
|
|
|
def test_list_endpoints_unauthenticated(self):
|
|
"""list_endpoints returns empty list when not authenticated."""
|
|
plugin = HarvesterPlugin()
|
|
assert plugin.list_endpoints() == []
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_list_endpoints_authenticated(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""list_endpoints returns the cluster API server URL."""
|
|
mock_api_client = MagicMock()
|
|
mock_api_client.configuration.host = "https://harvester.local:6443"
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
endpoints = plugin.list_endpoints()
|
|
assert endpoints == ["https://harvester.local:6443"]
|
|
|
|
|
|
class TestHarvesterPluginDetectArchitecture:
|
|
"""Tests for HarvesterPlugin.detect_architecture()."""
|
|
|
|
def test_detect_architecture_unauthenticated(self):
|
|
"""detect_architecture returns AMD64 when not authenticated."""
|
|
plugin = HarvesterPlugin()
|
|
arch = plugin.detect_architecture("https://harvester.local:6443")
|
|
assert arch == CpuArchitecture.AMD64
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_detect_architecture_amd64(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""detect_architecture returns AMD64 for amd64 nodes."""
|
|
mock_api_client = MagicMock()
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
mock_core_instance = MagicMock()
|
|
mock_core_cls.return_value = mock_core_instance
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
# Mock node list response
|
|
mock_node = MagicMock()
|
|
mock_node.status.node_info.architecture = "amd64"
|
|
mock_core_instance.list_node.return_value = MagicMock(items=[mock_node])
|
|
|
|
arch = plugin.detect_architecture("https://harvester.local:6443")
|
|
assert arch == CpuArchitecture.AMD64
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_detect_architecture_arm64(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""detect_architecture returns AARCH64 for arm64 nodes."""
|
|
mock_api_client = MagicMock()
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
mock_core_instance = MagicMock()
|
|
mock_core_cls.return_value = mock_core_instance
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.status.node_info.architecture = "arm64"
|
|
mock_core_instance.list_node.return_value = MagicMock(items=[mock_node])
|
|
|
|
arch = plugin.detect_architecture("https://harvester.local:6443")
|
|
assert arch == CpuArchitecture.AARCH64
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_detect_architecture_arm(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""detect_architecture returns ARM for arm nodes."""
|
|
mock_api_client = MagicMock()
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
mock_core_instance = MagicMock()
|
|
mock_core_cls.return_value = mock_core_instance
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.status.node_info.architecture = "arm"
|
|
mock_core_instance.list_node.return_value = MagicMock(items=[mock_node])
|
|
|
|
arch = plugin.detect_architecture("https://harvester.local:6443")
|
|
assert arch == CpuArchitecture.ARM
|
|
|
|
@patch(PATCH_CORE_API)
|
|
@patch(PATCH_CUSTOM_API)
|
|
@patch(PATCH_NEW_CLIENT)
|
|
def test_detect_architecture_api_error_defaults_amd64(
|
|
self, mock_new_client, mock_custom_cls, mock_core_cls
|
|
):
|
|
"""detect_architecture defaults to AMD64 on API errors."""
|
|
from kubernetes.client.rest import ApiException
|
|
|
|
mock_api_client = MagicMock()
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
mock_core_instance = MagicMock()
|
|
mock_core_cls.return_value = mock_core_instance
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
mock_core_instance.list_node.side_effect = ApiException(status=403)
|
|
|
|
arch = plugin.detect_architecture("https://harvester.local:6443")
|
|
assert arch == CpuArchitecture.AMD64
|
|
|
|
|
|
def _make_authenticated_plugin():
|
|
"""Create an authenticated plugin with mocked APIs.
|
|
|
|
Returns (plugin, mock_custom_api, mock_core_api) tuple.
|
|
"""
|
|
with patch(PATCH_NEW_CLIENT) as mock_new_client, \
|
|
patch(PATCH_CUSTOM_API) as mock_custom_cls, \
|
|
patch(PATCH_CORE_API) as mock_core_cls:
|
|
|
|
mock_api_client = MagicMock()
|
|
mock_api_client.configuration.host = "https://harvester.local:6443"
|
|
mock_new_client.return_value = mock_api_client
|
|
|
|
mock_custom_instance = MagicMock()
|
|
mock_custom_cls.return_value = mock_custom_instance
|
|
|
|
mock_core_instance = MagicMock()
|
|
mock_core_cls.return_value = mock_core_instance
|
|
|
|
plugin = HarvesterPlugin()
|
|
plugin.authenticate({"kubeconfig_path": "/path/to/kubeconfig"})
|
|
|
|
# Mock node for architecture detection
|
|
mock_node = MagicMock()
|
|
mock_node.status.node_info.architecture = "amd64"
|
|
mock_core_instance.list_node.return_value = MagicMock(items=[mock_node])
|
|
|
|
return plugin, mock_custom_instance, mock_core_instance
|
|
|
|
|
|
class TestHarvesterPluginDiscoverResources:
|
|
"""Tests for HarvesterPlugin.discover_resources()."""
|
|
|
|
def test_discover_vms(self):
|
|
"""discover_resources discovers virtual machines."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
|
|
mock_custom_api.list_cluster_custom_object.return_value = {
|
|
"items": [
|
|
{
|
|
"metadata": {
|
|
"name": "test-vm",
|
|
"namespace": "default",
|
|
"uid": "vm-uid-123",
|
|
"labels": {"app": "web"},
|
|
"annotations": {},
|
|
},
|
|
"spec": {
|
|
"running": True,
|
|
"template": {
|
|
"spec": {
|
|
"volumes": [
|
|
{"dataVolume": {"name": "test-disk"}},
|
|
],
|
|
"networks": [
|
|
{"multus": {"networkName": "vlan100"}},
|
|
],
|
|
}
|
|
},
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
progress_updates = []
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_virtualmachine"],
|
|
progress_callback=lambda p: progress_updates.append(p),
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
vm = result.resources[0]
|
|
assert vm.resource_type == "harvester_virtualmachine"
|
|
assert vm.name == "test-vm"
|
|
assert vm.unique_id == "vm-uid-123"
|
|
assert vm.provider == ProviderType.HARVESTER
|
|
assert vm.platform_category == PlatformCategory.HCI
|
|
assert vm.architecture == CpuArchitecture.AMD64
|
|
assert vm.attributes["running"] is True
|
|
assert "volume:test-disk" in vm.raw_references
|
|
assert "network:vlan100" in vm.raw_references
|
|
|
|
def test_discover_volumes(self):
|
|
"""discover_resources discovers data volumes."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
|
|
mock_custom_api.list_cluster_custom_object.return_value = {
|
|
"items": [
|
|
{
|
|
"metadata": {
|
|
"name": "test-disk",
|
|
"namespace": "default",
|
|
"uid": "vol-uid-456",
|
|
"labels": {},
|
|
},
|
|
"spec": {
|
|
"source": {"http": {"url": "https://images.example.com/disk.img"}},
|
|
"pvc": {"accessModes": ["ReadWriteOnce"], "resources": {"requests": {"storage": "10Gi"}}},
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_volume"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
vol = result.resources[0]
|
|
assert vol.resource_type == "harvester_volume"
|
|
assert vol.name == "test-disk"
|
|
assert vol.unique_id == "vol-uid-456"
|
|
|
|
def test_discover_images(self):
|
|
"""discover_resources discovers VM images."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
|
|
mock_custom_api.list_cluster_custom_object.return_value = {
|
|
"items": [
|
|
{
|
|
"metadata": {
|
|
"name": "ubuntu-22.04",
|
|
"namespace": "default",
|
|
"uid": "img-uid-789",
|
|
"labels": {"os": "linux"},
|
|
},
|
|
"spec": {
|
|
"displayName": "Ubuntu 22.04 LTS",
|
|
"url": "https://cloud-images.ubuntu.com/jammy/current/jammy-server-cloudimg-amd64.img",
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_image"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
img = result.resources[0]
|
|
assert img.resource_type == "harvester_image"
|
|
assert img.name == "ubuntu-22.04"
|
|
assert img.attributes["display_name"] == "Ubuntu 22.04 LTS"
|
|
assert "ubuntu.com" in img.attributes["url"]
|
|
|
|
def test_discover_networks(self):
|
|
"""discover_resources discovers network attachment definitions."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
|
|
mock_custom_api.list_cluster_custom_object.return_value = {
|
|
"items": [
|
|
{
|
|
"metadata": {
|
|
"name": "vlan100",
|
|
"namespace": "default",
|
|
"uid": "net-uid-abc",
|
|
"labels": {},
|
|
},
|
|
"spec": {
|
|
"config": '{"cniVersion":"0.3.1","name":"vlan100","type":"bridge"}',
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_network"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 1
|
|
net = result.resources[0]
|
|
assert net.resource_type == "harvester_network"
|
|
assert net.name == "vlan100"
|
|
assert "vlan100" in net.attributes["config"]
|
|
|
|
def test_discover_multiple_resource_types(self):
|
|
"""discover_resources handles multiple resource types in one call."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
|
|
# Return different items based on the CRD being queried
|
|
def mock_list_custom_object(group, version, plural):
|
|
if plural == "virtualmachines":
|
|
return {"items": [{"metadata": {"name": "vm1", "namespace": "default", "uid": "uid-1", "labels": {}, "annotations": {}}, "spec": {"running": True, "template": {"spec": {}}}}]}
|
|
elif plural == "network-attachment-definitions":
|
|
return {"items": [{"metadata": {"name": "net1", "namespace": "default", "uid": "uid-2", "labels": {}}, "spec": {"config": "{}"}}]}
|
|
return {"items": []}
|
|
|
|
mock_custom_api.list_cluster_custom_object.side_effect = mock_list_custom_object
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_virtualmachine", "harvester_network"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 2
|
|
types = {r.resource_type for r in result.resources}
|
|
assert types == {"harvester_virtualmachine", "harvester_network"}
|
|
|
|
def test_discover_resources_api_error(self):
|
|
"""discover_resources records errors when API calls fail."""
|
|
from kubernetes.client.rest import ApiException
|
|
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
mock_custom_api.list_cluster_custom_object.side_effect = ApiException(
|
|
status=403, reason="Forbidden"
|
|
)
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_virtualmachine"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.resources) == 0
|
|
assert len(result.errors) == 1
|
|
assert "403" in result.errors[0]
|
|
|
|
def test_discover_resources_progress_callback(self):
|
|
"""discover_resources invokes progress_callback correctly."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
mock_custom_api.list_cluster_custom_object.return_value = {"items": []}
|
|
|
|
progress_updates: list[ScanProgress] = []
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_virtualmachine", "harvester_volume"],
|
|
progress_callback=lambda p: progress_updates.append(p),
|
|
)
|
|
|
|
# Should have progress updates: one per resource type + final
|
|
assert len(progress_updates) == 3
|
|
assert progress_updates[0].current_resource_type == "harvester_virtualmachine"
|
|
assert progress_updates[0].total_resource_types == 2
|
|
assert progress_updates[-1].resource_types_completed == 2
|
|
|
|
def test_discover_resources_unknown_type_warning(self):
|
|
"""discover_resources warns about unknown resource types."""
|
|
plugin, mock_custom_api, _ = _make_authenticated_plugin()
|
|
|
|
result = plugin.discover_resources(
|
|
endpoints=["https://harvester.local:6443"],
|
|
resource_types=["harvester_unknown"],
|
|
progress_callback=lambda p: None,
|
|
)
|
|
|
|
assert len(result.warnings) == 1
|
|
assert "Unknown resource type" in result.warnings[0]
|
|
|
|
|
|
class TestHarvesterPluginVMReferences:
|
|
"""Tests for VM reference extraction."""
|
|
|
|
def test_extract_vm_references_data_volume(self):
|
|
"""Extracts dataVolume references from VM spec."""
|
|
spec = {
|
|
"template": {
|
|
"spec": {
|
|
"volumes": [
|
|
{"dataVolume": {"name": "my-disk"}},
|
|
],
|
|
"networks": [],
|
|
}
|
|
}
|
|
}
|
|
refs = HarvesterPlugin._extract_vm_references(spec)
|
|
assert refs == ["volume:my-disk"]
|
|
|
|
def test_extract_vm_references_pvc(self):
|
|
"""Extracts persistentVolumeClaim references from VM spec."""
|
|
spec = {
|
|
"template": {
|
|
"spec": {
|
|
"volumes": [
|
|
{"persistentVolumeClaim": {"claimName": "my-pvc"}},
|
|
],
|
|
"networks": [],
|
|
}
|
|
}
|
|
}
|
|
refs = HarvesterPlugin._extract_vm_references(spec)
|
|
assert refs == ["volume:my-pvc"]
|
|
|
|
def test_extract_vm_references_multus_network(self):
|
|
"""Extracts multus network references from VM spec."""
|
|
spec = {
|
|
"template": {
|
|
"spec": {
|
|
"volumes": [],
|
|
"networks": [
|
|
{"multus": {"networkName": "vlan200"}},
|
|
],
|
|
}
|
|
}
|
|
}
|
|
refs = HarvesterPlugin._extract_vm_references(spec)
|
|
assert refs == ["network:vlan200"]
|
|
|
|
def test_extract_vm_references_empty_spec(self):
|
|
"""Returns empty list for empty spec."""
|
|
refs = HarvesterPlugin._extract_vm_references({})
|
|
assert refs == []
|
|
|
|
def test_extract_vm_references_mixed(self):
|
|
"""Extracts both volume and network references."""
|
|
spec = {
|
|
"template": {
|
|
"spec": {
|
|
"volumes": [
|
|
{"dataVolume": {"name": "disk-1"}},
|
|
{"persistentVolumeClaim": {"claimName": "pvc-1"}},
|
|
],
|
|
"networks": [
|
|
{"multus": {"networkName": "mgmt-net"}},
|
|
{"multus": {"networkName": "data-net"}},
|
|
],
|
|
}
|
|
}
|
|
}
|
|
refs = HarvesterPlugin._extract_vm_references(spec)
|
|
assert refs == [
|
|
"volume:disk-1",
|
|
"volume:pvc-1",
|
|
"network:mgmt-net",
|
|
"network:data-net",
|
|
]
|