mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-05 19:51:52 +00:00
Daniel Barranquero
499 lines
20 KiB
Python
499 lines
20 KiB
Python
"""Tests for OpenStack Compute service."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from openstack import exceptions as openstack_exceptions
|
|
|
|
from prowler.providers.openstack.services.compute.compute_service import (
|
|
Compute,
|
|
ComputeInstance,
|
|
)
|
|
from tests.providers.openstack.openstack_fixtures import (
|
|
OPENSTACK_PROJECT_ID,
|
|
OPENSTACK_REGION,
|
|
set_mocked_openstack_provider,
|
|
)
|
|
|
|
|
|
class TestComputeService:
|
|
"""Test suite for Compute service."""
|
|
|
|
def test_compute_service_initialization(self):
|
|
"""Test Compute service initializes correctly."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
with patch.object(Compute, "_list_instances", return_value=[]) as mock_list:
|
|
compute = Compute(provider)
|
|
|
|
assert compute.service_name == "Compute"
|
|
assert compute.provider == provider
|
|
assert compute.connection == provider.connection
|
|
assert compute.regional_connections == provider.regional_connections
|
|
assert compute.audited_regions == [OPENSTACK_REGION]
|
|
assert compute.region == OPENSTACK_REGION
|
|
assert compute.project_id == OPENSTACK_PROJECT_ID
|
|
assert compute.instances == []
|
|
mock_list.assert_called_once()
|
|
|
|
def test_compute_list_instances_success(self):
|
|
"""Test listing compute instances successfully."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
mock_server1 = MagicMock()
|
|
mock_server1.id = "instance-1"
|
|
mock_server1.name = "Instance One"
|
|
mock_server1.status = "ACTIVE"
|
|
mock_server1.flavor = {"id": "flavor-1"}
|
|
mock_server1.security_groups = [{"name": "default"}]
|
|
mock_server1.is_locked = True
|
|
mock_server1.locked_reason = "maintenance"
|
|
mock_server1.key_name = "my-keypair"
|
|
mock_server1.user_id = "user-123"
|
|
mock_server1.access_ipv4 = "203.0.113.10"
|
|
mock_server1.access_ipv6 = "2001:db8::1"
|
|
mock_server1.public_v4 = "203.0.113.10"
|
|
mock_server1.public_v6 = ""
|
|
mock_server1.private_v4 = "10.0.0.5"
|
|
mock_server1.private_v6 = ""
|
|
mock_server1.addresses = {
|
|
"private": [{"version": 4, "addr": "10.0.0.5"}],
|
|
"public": [{"version": 4, "addr": "203.0.113.10"}],
|
|
}
|
|
mock_server1.has_config_drive = True
|
|
mock_server1.metadata = {"environment": "production"}
|
|
mock_server1.user_data = "#!/bin/bash\necho hello"
|
|
mock_server1.trusted_image_certificates = ["cert-123"]
|
|
|
|
mock_server2 = MagicMock()
|
|
mock_server2.id = "instance-2"
|
|
mock_server2.name = "Instance Two"
|
|
mock_server2.status = "SHUTOFF"
|
|
mock_server2.flavor = {"id": "flavor-2"}
|
|
mock_server2.security_groups = [{"name": "web"}, {"name": "db"}]
|
|
mock_server2.is_locked = False
|
|
mock_server2.locked_reason = ""
|
|
mock_server2.key_name = ""
|
|
mock_server2.user_id = "user-456"
|
|
mock_server2.access_ipv4 = ""
|
|
mock_server2.access_ipv6 = ""
|
|
mock_server2.public_v4 = ""
|
|
mock_server2.public_v6 = ""
|
|
mock_server2.private_v4 = "10.0.0.10"
|
|
mock_server2.private_v6 = ""
|
|
mock_server2.addresses = {"private": [{"version": 4, "addr": "10.0.0.10"}]}
|
|
mock_server2.has_config_drive = False
|
|
mock_server2.metadata = {}
|
|
mock_server2.user_data = ""
|
|
mock_server2.trusted_image_certificates = []
|
|
|
|
provider.connection.compute.servers.return_value = [
|
|
mock_server1,
|
|
mock_server2,
|
|
]
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert len(compute.instances) == 2
|
|
assert isinstance(compute.instances[0], ComputeInstance)
|
|
assert compute.instances[0].id == "instance-1"
|
|
assert compute.instances[0].name == "Instance One"
|
|
assert compute.instances[0].status == "ACTIVE"
|
|
assert compute.instances[0].flavor_id == "flavor-1"
|
|
assert compute.instances[0].security_groups == ["default"]
|
|
assert compute.instances[0].region == OPENSTACK_REGION
|
|
assert compute.instances[0].project_id == OPENSTACK_PROJECT_ID
|
|
assert compute.instances[0].is_locked is True
|
|
assert compute.instances[0].locked_reason == "maintenance"
|
|
assert compute.instances[0].key_name == "my-keypair"
|
|
assert compute.instances[0].user_id == "user-123"
|
|
assert compute.instances[0].access_ipv4 == "203.0.113.10"
|
|
assert compute.instances[0].access_ipv6 == "2001:db8::1"
|
|
assert compute.instances[0].public_v4 == "203.0.113.10"
|
|
assert compute.instances[0].private_v4 == "10.0.0.5"
|
|
assert compute.instances[0].networks == {
|
|
"private": ["10.0.0.5"],
|
|
"public": ["203.0.113.10"],
|
|
}
|
|
assert compute.instances[0].has_config_drive is True
|
|
assert compute.instances[0].metadata == {"environment": "production"}
|
|
assert compute.instances[0].user_data == "#!/bin/bash\necho hello"
|
|
assert compute.instances[0].trusted_image_certificates == ["cert-123"]
|
|
|
|
assert compute.instances[1].security_groups == ["web", "db"]
|
|
assert compute.instances[1].is_locked is False
|
|
assert compute.instances[1].key_name == ""
|
|
assert compute.instances[1].trusted_image_certificates == []
|
|
|
|
def test_compute_list_instances_empty(self):
|
|
"""Test listing instances when none exist."""
|
|
provider = set_mocked_openstack_provider()
|
|
provider.connection.compute.servers.return_value = []
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert compute.instances == []
|
|
|
|
def test_compute_list_instances_missing_attributes(self):
|
|
"""Test listing instances with missing attributes."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
mock_server = MagicMock()
|
|
mock_server.id = "instance-1"
|
|
del mock_server.name
|
|
del mock_server.status
|
|
del mock_server.flavor
|
|
del mock_server.security_groups
|
|
del mock_server.is_locked
|
|
del mock_server.locked_reason
|
|
del mock_server.key_name
|
|
del mock_server.user_id
|
|
del mock_server.access_ipv4
|
|
del mock_server.access_ipv6
|
|
del mock_server.public_v4
|
|
del mock_server.public_v6
|
|
del mock_server.private_v4
|
|
del mock_server.private_v6
|
|
del mock_server.addresses
|
|
del mock_server.has_config_drive
|
|
del mock_server.metadata
|
|
del mock_server.user_data
|
|
del mock_server.trusted_image_certificates
|
|
|
|
provider.connection.compute.servers.return_value = [mock_server]
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert len(compute.instances) == 1
|
|
assert compute.instances[0].id == "instance-1"
|
|
assert compute.instances[0].name == ""
|
|
assert compute.instances[0].status == ""
|
|
assert compute.instances[0].flavor_id == ""
|
|
assert compute.instances[0].security_groups == []
|
|
assert compute.instances[0].is_locked is False
|
|
assert compute.instances[0].locked_reason == ""
|
|
assert compute.instances[0].key_name == ""
|
|
assert compute.instances[0].user_id == ""
|
|
assert compute.instances[0].access_ipv4 == ""
|
|
assert compute.instances[0].access_ipv6 == ""
|
|
assert compute.instances[0].public_v4 == ""
|
|
assert compute.instances[0].public_v6 == ""
|
|
assert compute.instances[0].private_v4 == ""
|
|
assert compute.instances[0].private_v6 == ""
|
|
assert compute.instances[0].networks == {}
|
|
assert compute.instances[0].has_config_drive is False
|
|
assert compute.instances[0].metadata == {}
|
|
assert compute.instances[0].user_data == ""
|
|
assert compute.instances[0].trusted_image_certificates == []
|
|
|
|
def test_compute_list_instances_sdk_exception(self):
|
|
"""Test handling SDKException when listing instances."""
|
|
provider = set_mocked_openstack_provider()
|
|
provider.connection.compute.servers.side_effect = (
|
|
openstack_exceptions.SDKException("API error")
|
|
)
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert compute.instances == []
|
|
|
|
def test_compute_list_instances_generic_exception(self):
|
|
"""Test handling generic exception when listing instances."""
|
|
provider = set_mocked_openstack_provider()
|
|
provider.connection.compute.servers.side_effect = Exception("Unexpected error")
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert compute.instances == []
|
|
|
|
def test_compute_list_instances_iterator_exception(self):
|
|
"""Test listing instances when iterator fails mid-stream."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
def failing_iterator():
|
|
mock_server = MagicMock()
|
|
mock_server.id = "instance-1"
|
|
mock_server.name = "Instance One"
|
|
mock_server.status = "ACTIVE"
|
|
mock_server.flavor = {"id": "flavor-1"}
|
|
mock_server.security_groups = [{"name": "default"}]
|
|
mock_server.is_locked = False
|
|
mock_server.locked_reason = ""
|
|
mock_server.key_name = ""
|
|
mock_server.user_id = ""
|
|
mock_server.access_ipv4 = ""
|
|
mock_server.access_ipv6 = ""
|
|
mock_server.public_v4 = ""
|
|
mock_server.public_v6 = ""
|
|
mock_server.private_v4 = ""
|
|
mock_server.private_v6 = ""
|
|
mock_server.addresses = {}
|
|
mock_server.has_config_drive = False
|
|
mock_server.metadata = {}
|
|
mock_server.user_data = ""
|
|
mock_server.trusted_image_certificates = []
|
|
yield mock_server
|
|
raise Exception("Iterator failed")
|
|
|
|
provider.connection.compute.servers.return_value = failing_iterator()
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert len(compute.instances) == 1
|
|
assert compute.instances[0].id == "instance-1"
|
|
assert compute.instances[0].name == "Instance One"
|
|
|
|
def test_compute_instance_dataclass_attributes(self):
|
|
"""Test ComputeInstance dataclass has all required attributes."""
|
|
instance = ComputeInstance(
|
|
id="instance-1",
|
|
name="Instance One",
|
|
status="ACTIVE",
|
|
flavor_id="flavor-1",
|
|
security_groups=["default"],
|
|
region="RegionOne",
|
|
project_id="project-1",
|
|
is_locked=True,
|
|
locked_reason="maintenance",
|
|
key_name="my-keypair",
|
|
user_id="user-123",
|
|
access_ipv4="203.0.113.10",
|
|
access_ipv6="2001:db8::1",
|
|
public_v4="203.0.113.10",
|
|
public_v6="",
|
|
private_v4="10.0.0.5",
|
|
private_v6="",
|
|
networks={
|
|
"private": ["10.0.0.5"]
|
|
}, # Note: This is the processed dict, not addresses
|
|
has_config_drive=True,
|
|
metadata={"environment": "production"},
|
|
user_data="#!/bin/bash\necho hello",
|
|
trusted_image_certificates=["cert-123"],
|
|
)
|
|
|
|
assert instance.id == "instance-1"
|
|
assert instance.name == "Instance One"
|
|
assert instance.status == "ACTIVE"
|
|
assert instance.flavor_id == "flavor-1"
|
|
assert instance.security_groups == ["default"]
|
|
assert instance.region == "RegionOne"
|
|
assert instance.project_id == "project-1"
|
|
assert instance.is_locked is True
|
|
assert instance.locked_reason == "maintenance"
|
|
assert instance.key_name == "my-keypair"
|
|
assert instance.user_id == "user-123"
|
|
assert instance.access_ipv4 == "203.0.113.10"
|
|
assert instance.access_ipv6 == "2001:db8::1"
|
|
assert instance.public_v4 == "203.0.113.10"
|
|
assert instance.public_v6 == ""
|
|
assert instance.private_v4 == "10.0.0.5"
|
|
assert instance.private_v6 == ""
|
|
assert instance.networks == {"private": ["10.0.0.5"]}
|
|
assert instance.has_config_drive is True
|
|
assert instance.metadata == {"environment": "production"}
|
|
assert instance.user_data == "#!/bin/bash\necho hello"
|
|
assert instance.trusted_image_certificates == ["cert-123"]
|
|
|
|
def test_compute_service_inherits_from_base(self):
|
|
"""Test Compute service inherits from OpenStackService."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
with patch.object(Compute, "_list_instances", return_value=[]):
|
|
compute = Compute(provider)
|
|
|
|
assert hasattr(compute, "service_name")
|
|
assert hasattr(compute, "provider")
|
|
assert hasattr(compute, "connection")
|
|
assert hasattr(compute, "regional_connections")
|
|
assert hasattr(compute, "audited_regions")
|
|
assert hasattr(compute, "session")
|
|
assert hasattr(compute, "region")
|
|
assert hasattr(compute, "project_id")
|
|
assert hasattr(compute, "identity")
|
|
assert hasattr(compute, "audit_config")
|
|
assert hasattr(compute, "fixer_config")
|
|
|
|
def test_compute_list_instances_with_none_addresses(self):
|
|
"""Test listing instances when addresses attribute is None."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
mock_server = MagicMock()
|
|
mock_server.id = "instance-1"
|
|
mock_server.name = "Instance With None Addresses"
|
|
mock_server.status = "ACTIVE"
|
|
mock_server.flavor = {"id": "flavor-1"}
|
|
mock_server.security_groups = [{"name": "default"}]
|
|
mock_server.is_locked = False
|
|
mock_server.locked_reason = ""
|
|
mock_server.key_name = "test-key"
|
|
mock_server.user_id = "user-123"
|
|
mock_server.access_ipv4 = ""
|
|
mock_server.access_ipv6 = ""
|
|
mock_server.public_v4 = ""
|
|
mock_server.public_v6 = ""
|
|
mock_server.private_v4 = ""
|
|
mock_server.private_v6 = ""
|
|
mock_server.addresses = None # This is the key test case
|
|
mock_server.has_config_drive = False
|
|
mock_server.metadata = {}
|
|
mock_server.user_data = ""
|
|
mock_server.trusted_image_certificates = []
|
|
|
|
provider.connection.compute.servers.return_value = [mock_server]
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert len(compute.instances) == 1
|
|
assert compute.instances[0].id == "instance-1"
|
|
assert compute.instances[0].networks == {} # Should default to empty dict
|
|
|
|
def test_compute_list_instances_multi_region(self):
|
|
"""Test listing instances across multiple regions."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
# Create two mock connections for two regions
|
|
mock_conn_uk1 = MagicMock()
|
|
mock_conn_de1 = MagicMock()
|
|
|
|
# Set up regional connections
|
|
provider.regional_connections = {"UK1": mock_conn_uk1, "DE1": mock_conn_de1}
|
|
|
|
mock_server_uk = MagicMock()
|
|
mock_server_uk.id = "instance-uk"
|
|
mock_server_uk.name = "Instance UK"
|
|
mock_server_uk.status = "ACTIVE"
|
|
mock_server_uk.flavor = {"id": "flavor-1"}
|
|
mock_server_uk.security_groups = [{"name": "default"}]
|
|
mock_server_uk.is_locked = False
|
|
mock_server_uk.locked_reason = ""
|
|
mock_server_uk.key_name = ""
|
|
mock_server_uk.user_id = ""
|
|
mock_server_uk.access_ipv4 = ""
|
|
mock_server_uk.access_ipv6 = ""
|
|
mock_server_uk.public_v4 = ""
|
|
mock_server_uk.public_v6 = ""
|
|
mock_server_uk.private_v4 = "10.0.0.1"
|
|
mock_server_uk.private_v6 = ""
|
|
mock_server_uk.addresses = {"private": [{"version": 4, "addr": "10.0.0.1"}]}
|
|
mock_server_uk.has_config_drive = False
|
|
mock_server_uk.metadata = {}
|
|
mock_server_uk.user_data = ""
|
|
mock_server_uk.trusted_image_certificates = []
|
|
|
|
mock_server_de = MagicMock()
|
|
mock_server_de.id = "instance-de"
|
|
mock_server_de.name = "Instance DE"
|
|
mock_server_de.status = "ACTIVE"
|
|
mock_server_de.flavor = {"id": "flavor-2"}
|
|
mock_server_de.security_groups = [{"name": "default"}]
|
|
mock_server_de.is_locked = False
|
|
mock_server_de.locked_reason = ""
|
|
mock_server_de.key_name = ""
|
|
mock_server_de.user_id = ""
|
|
mock_server_de.access_ipv4 = ""
|
|
mock_server_de.access_ipv6 = ""
|
|
mock_server_de.public_v4 = ""
|
|
mock_server_de.public_v6 = ""
|
|
mock_server_de.private_v4 = "10.0.0.2"
|
|
mock_server_de.private_v6 = ""
|
|
mock_server_de.addresses = {"private": [{"version": 4, "addr": "10.0.0.2"}]}
|
|
mock_server_de.has_config_drive = False
|
|
mock_server_de.metadata = {}
|
|
mock_server_de.user_data = ""
|
|
mock_server_de.trusted_image_certificates = []
|
|
|
|
mock_conn_uk1.compute.servers.return_value = [mock_server_uk]
|
|
mock_conn_de1.compute.servers.return_value = [mock_server_de]
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert len(compute.instances) == 2
|
|
# Verify instances have correct region tags
|
|
uk_instance = next(i for i in compute.instances if i.id == "instance-uk")
|
|
de_instance = next(i for i in compute.instances if i.id == "instance-de")
|
|
assert uk_instance.region == "UK1"
|
|
assert de_instance.region == "DE1"
|
|
|
|
def test_compute_list_instances_multi_region_partial_failure(self):
|
|
"""Test that a failing region doesn't prevent other regions from being listed."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
mock_conn_ok = MagicMock()
|
|
mock_conn_fail = MagicMock()
|
|
|
|
provider.regional_connections = {"UK1": mock_conn_ok, "DE1": mock_conn_fail}
|
|
|
|
mock_server = MagicMock()
|
|
mock_server.id = "instance-uk"
|
|
mock_server.name = "Instance UK"
|
|
mock_server.status = "ACTIVE"
|
|
mock_server.flavor = {"id": "flavor-1"}
|
|
mock_server.security_groups = [{"name": "default"}]
|
|
mock_server.is_locked = False
|
|
mock_server.locked_reason = ""
|
|
mock_server.key_name = ""
|
|
mock_server.user_id = ""
|
|
mock_server.access_ipv4 = ""
|
|
mock_server.access_ipv6 = ""
|
|
mock_server.public_v4 = ""
|
|
mock_server.public_v6 = ""
|
|
mock_server.private_v4 = "10.0.0.1"
|
|
mock_server.private_v6 = ""
|
|
mock_server.addresses = {}
|
|
mock_server.has_config_drive = False
|
|
mock_server.metadata = {}
|
|
mock_server.user_data = ""
|
|
mock_server.trusted_image_certificates = []
|
|
|
|
mock_conn_ok.compute.servers.return_value = [mock_server]
|
|
mock_conn_fail.compute.servers.side_effect = openstack_exceptions.SDKException(
|
|
"API error in DE1"
|
|
)
|
|
|
|
compute = Compute(provider)
|
|
|
|
# Should have the instance from UK1, DE1 failure is logged but doesn't crash
|
|
assert len(compute.instances) == 1
|
|
assert compute.instances[0].id == "instance-uk"
|
|
assert compute.instances[0].region == "UK1"
|
|
|
|
def test_compute_list_instances_multi_region_one_empty(self):
|
|
"""Test multi-region where one region has instances and the other is empty."""
|
|
provider = set_mocked_openstack_provider()
|
|
|
|
mock_conn_uk1 = MagicMock()
|
|
mock_conn_de1 = MagicMock()
|
|
|
|
provider.regional_connections = {"UK1": mock_conn_uk1, "DE1": mock_conn_de1}
|
|
|
|
mock_server = MagicMock()
|
|
mock_server.id = "instance-uk"
|
|
mock_server.name = "Instance UK"
|
|
mock_server.status = "ACTIVE"
|
|
mock_server.flavor = {"id": "flavor-1"}
|
|
mock_server.security_groups = [{"name": "default"}]
|
|
mock_server.is_locked = False
|
|
mock_server.locked_reason = ""
|
|
mock_server.key_name = ""
|
|
mock_server.user_id = ""
|
|
mock_server.access_ipv4 = ""
|
|
mock_server.access_ipv6 = ""
|
|
mock_server.public_v4 = ""
|
|
mock_server.public_v6 = ""
|
|
mock_server.private_v4 = "10.0.0.1"
|
|
mock_server.private_v6 = ""
|
|
mock_server.addresses = {}
|
|
mock_server.has_config_drive = False
|
|
mock_server.metadata = {}
|
|
mock_server.user_data = ""
|
|
mock_server.trusted_image_certificates = []
|
|
|
|
mock_conn_uk1.compute.servers.return_value = [mock_server]
|
|
mock_conn_de1.compute.servers.return_value = [] # Empty region
|
|
|
|
compute = Compute(provider)
|
|
|
|
assert len(compute.instances) == 1
|
|
assert compute.instances[0].id == "instance-uk"
|
|
assert compute.instances[0].region == "UK1"
|