Add wrapper classes for novaclient objects
Add Server, Hypervisor, Flavor, Aggregate, and Service wrapper classes to provide cleaner interfaces to novaclient objects. These wrappers: - Use dataclasses to provide basic method like __eq__ and __repr__ - Provide clean property access for commonly used attributes - Handle OpenStack extension attributes (OS-EXT-*) transparently - Avoid getattr() when direct attribute access is possible Future commits will use these classes in the nova_helper instead of returning novaclient objects. This will allow to switch novaclient usage to openstacksdk without further changes to the watcher code base. Generated-By: claude-code (claude-sonnet-4.5) Change-Id: Icee1398e07786a1b9ff7da7aa7fa7e13193c143b Signed-off-by: jgilaber <jgilaber@redhat.com>
This commit is contained in:
jgilaber
@@ -16,6 +16,7 @@
|
||||
# limitations under the License.
|
||||
#
|
||||
|
||||
import dataclasses as dc
|
||||
import functools
|
||||
import time
|
||||
|
||||
@@ -55,6 +56,252 @@ def nova_retries(call):
|
||||
return wrapper
|
||||
|
||||
|
||||
@dc.dataclass(frozen=True)
|
||||
class Server:
|
||||
"""Pure dataclass for server data.
|
||||
|
||||
Extracted from novaclient Server object with all extended attributes
|
||||
resolved at construction time.
|
||||
"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
created: str
|
||||
host: str | None
|
||||
vm_state: str | None
|
||||
task_state: str | None
|
||||
power_state: int | None
|
||||
status: str
|
||||
flavor: dict
|
||||
tenant_id: str
|
||||
locked: bool
|
||||
metadata: dict
|
||||
availability_zone: str | None
|
||||
pinned_availability_zone: str | None
|
||||
|
||||
@classmethod
|
||||
def from_novaclient(cls, nova_server):
|
||||
"""Create a Server dataclass from a novaclient Server object.
|
||||
|
||||
:param nova_server: novaclient servers.Server object
|
||||
:returns: Server dataclass instance
|
||||
"""
|
||||
server_dict = nova_server.to_dict()
|
||||
|
||||
return cls(
|
||||
id=nova_server.id,
|
||||
name=nova_server.name,
|
||||
created=nova_server.created,
|
||||
host=server_dict.get('OS-EXT-SRV-ATTR:host'),
|
||||
vm_state=server_dict.get('OS-EXT-STS:vm_state'),
|
||||
task_state=server_dict.get('OS-EXT-STS:task_state'),
|
||||
power_state=server_dict.get('OS-EXT-STS:power_state'),
|
||||
status=nova_server.status,
|
||||
flavor=nova_server.flavor,
|
||||
tenant_id=nova_server.tenant_id,
|
||||
locked=nova_server.locked,
|
||||
metadata=nova_server.metadata,
|
||||
availability_zone=server_dict.get('OS-EXT-AZ:availability_zone'),
|
||||
pinned_availability_zone=server_dict.get(
|
||||
'pinned_availability_zone'
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@dc.dataclass(frozen=True)
|
||||
class Hypervisor:
|
||||
"""Pure dataclass for hypervisor data.
|
||||
|
||||
Extracted from novaclient Hypervisor object with all extended attributes
|
||||
resolved at construction time.
|
||||
"""
|
||||
|
||||
id: str
|
||||
hypervisor_hostname: str
|
||||
hypervisor_type: str
|
||||
state: str
|
||||
status: str
|
||||
vcpus: int | None
|
||||
vcpus_used: int | None
|
||||
memory_mb: int | None
|
||||
memory_mb_used: int | None
|
||||
local_gb: int | None
|
||||
local_gb_used: int | None
|
||||
service_host: str | None
|
||||
service_id: str | None
|
||||
service_disabled_reason: str | None
|
||||
servers: list | None
|
||||
|
||||
@classmethod
|
||||
def from_novaclient(cls, nova_hypervisor):
|
||||
"""Create a Hypervisor dataclass from a novaclient Hypervisor object.
|
||||
|
||||
:param nova_hypervisor: novaclient hypervisors.Hypervisor object
|
||||
:returns: Hypervisor dataclass instance
|
||||
"""
|
||||
hypervisor_dict = nova_hypervisor.to_dict()
|
||||
service = hypervisor_dict.get('service')
|
||||
service_host = None
|
||||
service_id = None
|
||||
service_disabled_reason = None
|
||||
if isinstance(service, dict):
|
||||
service_host = service.get('host')
|
||||
service_id = service.get('id')
|
||||
service_disabled_reason = service.get('disabled_reason')
|
||||
|
||||
servers = hypervisor_dict.get('servers', [])
|
||||
|
||||
return cls(
|
||||
id=nova_hypervisor.id,
|
||||
hypervisor_hostname=nova_hypervisor.hypervisor_hostname,
|
||||
hypervisor_type=nova_hypervisor.hypervisor_type,
|
||||
state=nova_hypervisor.state,
|
||||
status=nova_hypervisor.status,
|
||||
vcpus=hypervisor_dict.get('vcpus'),
|
||||
vcpus_used=hypervisor_dict.get('vcpus_used'),
|
||||
memory_mb=hypervisor_dict.get('memory_mb'),
|
||||
memory_mb_used=hypervisor_dict.get('memory_mb_used'),
|
||||
local_gb=hypervisor_dict.get('local_gb'),
|
||||
local_gb_used=hypervisor_dict.get('local_gb_used'),
|
||||
service_host=service_host,
|
||||
service_id=service_id,
|
||||
service_disabled_reason=service_disabled_reason,
|
||||
servers=servers,
|
||||
)
|
||||
|
||||
|
||||
@dc.dataclass(frozen=True)
|
||||
class Flavor:
|
||||
"""Pure dataclass for flavor data.
|
||||
|
||||
Extracted from novaclient Flavor object with all attributes
|
||||
resolved at construction time.
|
||||
"""
|
||||
|
||||
id: str
|
||||
flavor_name: str
|
||||
vcpus: int
|
||||
ram: int
|
||||
disk: int
|
||||
ephemeral: int
|
||||
swap: int
|
||||
is_public: bool
|
||||
extra_specs: dict
|
||||
|
||||
@classmethod
|
||||
def from_novaclient(cls, nova_flavor):
|
||||
"""Create a Flavor dataclass from a novaclient Flavor object.
|
||||
|
||||
:param nova_flavor: novaclient flavors.Flavor object
|
||||
:returns: Flavor dataclass instance
|
||||
"""
|
||||
swap = nova_flavor.swap
|
||||
if swap == "":
|
||||
swap = 0
|
||||
|
||||
flavor_dict = nova_flavor.to_dict()
|
||||
|
||||
return cls(
|
||||
id=nova_flavor.id,
|
||||
flavor_name=nova_flavor.name,
|
||||
vcpus=nova_flavor.vcpus,
|
||||
ram=nova_flavor.ram,
|
||||
disk=nova_flavor.disk,
|
||||
ephemeral=nova_flavor.ephemeral,
|
||||
swap=swap,
|
||||
is_public=nova_flavor.is_public,
|
||||
extra_specs=flavor_dict.get('extra_specs', {})
|
||||
)
|
||||
|
||||
|
||||
@dc.dataclass(frozen=True)
|
||||
class Aggregate:
|
||||
"""Pure dataclass for aggregate data.
|
||||
|
||||
Extracted from novaclient Aggregate object with all attributes
|
||||
resolved at construction time.
|
||||
"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
availability_zone: str | None
|
||||
hosts: list
|
||||
metadata: dict
|
||||
|
||||
@classmethod
|
||||
def from_novaclient(cls, nova_aggregate):
|
||||
"""Create an Aggregate dataclass from a novaclient Aggregate object.
|
||||
|
||||
:param nova_aggregate: novaclient aggregates.Aggregate object
|
||||
:returns: Aggregate dataclass instance
|
||||
"""
|
||||
return cls(
|
||||
id=nova_aggregate.id,
|
||||
name=nova_aggregate.name,
|
||||
availability_zone=nova_aggregate.availability_zone,
|
||||
hosts=nova_aggregate.hosts,
|
||||
metadata=nova_aggregate.metadata,
|
||||
)
|
||||
|
||||
|
||||
@dc.dataclass(frozen=True)
|
||||
class Service:
|
||||
"""Pure dataclass for service data.
|
||||
|
||||
Extracted from novaclient Service object with all attributes
|
||||
resolved at construction time.
|
||||
"""
|
||||
|
||||
id: str
|
||||
binary: str
|
||||
host: str
|
||||
zone: str
|
||||
status: str
|
||||
state: str
|
||||
updated_at: str | None
|
||||
disabled_reason: str | None
|
||||
|
||||
@classmethod
|
||||
def from_novaclient(cls, nova_service):
|
||||
"""Create a Service dataclass from a novaclient Service object.
|
||||
|
||||
:param nova_service: novaclient services.Service object
|
||||
:returns: Service dataclass instance
|
||||
"""
|
||||
return cls(
|
||||
id=nova_service.id,
|
||||
binary=nova_service.binary,
|
||||
host=nova_service.host,
|
||||
zone=nova_service.zone,
|
||||
status=nova_service.status,
|
||||
state=nova_service.state,
|
||||
updated_at=nova_service.updated_at,
|
||||
disabled_reason=nova_service.disabled_reason,
|
||||
)
|
||||
|
||||
|
||||
@dc.dataclass(frozen=True)
|
||||
class ServerMigration:
|
||||
"""Pure dataclass for server migration data.
|
||||
|
||||
Extracted from novaclient ServerMigration object with all attributes
|
||||
resolved at construction time.
|
||||
"""
|
||||
|
||||
id: str
|
||||
|
||||
@classmethod
|
||||
def from_novaclient(cls, nova_migration):
|
||||
"""Create a ServerMigration from a novaclient ServerMigration.
|
||||
|
||||
:param nova_migration: novaclient server_migrations.ServerMigration
|
||||
:returns: ServerMigration dataclass instance
|
||||
"""
|
||||
return cls(
|
||||
id=nova_migration.id,
|
||||
)
|
||||
|
||||
|
||||
class NovaHelper:
|
||||
|
||||
def __init__(self, osc=None):
|
||||
|
||||
@@ -1257,3 +1257,628 @@ class TestNovaRetries(base.TestCase):
|
||||
self.assertEqual(2, call_count['count'])
|
||||
# Should have slept once
|
||||
self.assertEqual(1, self.mock_sleep.call_count)
|
||||
|
||||
|
||||
class TestServerWrapper(base.TestCase):
|
||||
"""Test suite for the Server dataclass."""
|
||||
|
||||
@staticmethod
|
||||
def create_nova_server(server_id, **kwargs):
|
||||
"""Create a real novaclient Server object.
|
||||
|
||||
:param server_id: server UUID
|
||||
:param kwargs: additional server attributes
|
||||
:returns: novaclient.v2.servers.Server object
|
||||
"""
|
||||
server_info = {
|
||||
'id': server_id,
|
||||
'name': kwargs.pop('name', 'test-server'),
|
||||
'status': kwargs.pop('status', 'ACTIVE'),
|
||||
'created': kwargs.pop('created', '2026-01-09T12:00:00Z'),
|
||||
'tenant_id': kwargs.pop('tenant_id', 'test-tenant-id'),
|
||||
'locked': kwargs.pop('locked', False),
|
||||
'metadata': kwargs.pop('metadata', {}),
|
||||
'flavor': kwargs.pop('flavor', {'id': 'flavor-1'}),
|
||||
'pinned_availability_zone': kwargs.pop(
|
||||
'pinned_availability_zone', None),
|
||||
}
|
||||
server_info.update(kwargs)
|
||||
return servers.Server(servers.ServerManager, info=server_info)
|
||||
|
||||
def test_server_basic_properties(self):
|
||||
"""Test basic Server dataclass properties."""
|
||||
server_id = utils.generate_uuid()
|
||||
nova_server = self.create_nova_server(
|
||||
server_id,
|
||||
name='my-server',
|
||||
status='ACTIVE',
|
||||
created='2026-01-01T00:00:00Z',
|
||||
tenant_id='tenant-123',
|
||||
locked=True,
|
||||
metadata={'key': 'value'},
|
||||
pinned_availability_zone='az1'
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Server.from_novaclient(nova_server)
|
||||
|
||||
self.assertEqual(server_id, wrapped.id)
|
||||
self.assertEqual('my-server', wrapped.name)
|
||||
self.assertEqual('ACTIVE', wrapped.status)
|
||||
self.assertEqual('2026-01-01T00:00:00Z', wrapped.created)
|
||||
self.assertEqual('tenant-123', wrapped.tenant_id)
|
||||
self.assertTrue(wrapped.locked)
|
||||
self.assertEqual({'key': 'value'}, wrapped.metadata)
|
||||
self.assertEqual('az1', wrapped.pinned_availability_zone)
|
||||
|
||||
def test_server_extended_attributes(self):
|
||||
"""Test Server dataclass extended attributes."""
|
||||
server_id = utils.generate_uuid()
|
||||
nova_server = self.create_nova_server(
|
||||
server_id,
|
||||
**{
|
||||
'OS-EXT-SRV-ATTR:host': 'compute-1',
|
||||
'OS-EXT-STS:vm_state': 'active',
|
||||
'OS-EXT-STS:task_state': None,
|
||||
'OS-EXT-STS:power_state': 1,
|
||||
'OS-EXT-AZ:availability_zone': 'nova',
|
||||
}
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Server.from_novaclient(nova_server)
|
||||
|
||||
self.assertEqual('compute-1', wrapped.host)
|
||||
self.assertEqual('active', wrapped.vm_state)
|
||||
self.assertIsNone(wrapped.task_state)
|
||||
self.assertEqual(1, wrapped.power_state)
|
||||
self.assertEqual('nova', wrapped.availability_zone)
|
||||
|
||||
def test_server_flavor(self):
|
||||
"""Test Server dataclass flavor property."""
|
||||
server_id = utils.generate_uuid()
|
||||
|
||||
nova_server = self.create_nova_server(
|
||||
server_id,
|
||||
flavor={'id': 'flavor-123', 'name': 'm1.small'}
|
||||
)
|
||||
wrapped = nova_helper.Server.from_novaclient(nova_server)
|
||||
self.assertEqual({'id': 'flavor-123', 'name': 'm1.small'},
|
||||
wrapped.flavor)
|
||||
|
||||
def test_server_equality(self):
|
||||
"""Test Server dataclass equality comparison."""
|
||||
server_id1 = utils.generate_uuid()
|
||||
server_id2 = utils.generate_uuid()
|
||||
|
||||
server1a = nova_helper.Server.from_novaclient(
|
||||
self.create_nova_server(server_id1)
|
||||
)
|
||||
server1b = nova_helper.Server.from_novaclient(
|
||||
self.create_nova_server(server_id1)
|
||||
)
|
||||
server2 = nova_helper.Server.from_novaclient(
|
||||
self.create_nova_server(server_id2)
|
||||
)
|
||||
|
||||
# Same ID and attributes should be equal
|
||||
self.assertEqual(server1a, server1b)
|
||||
|
||||
# Different ID should not be equal
|
||||
self.assertNotEqual(server1a, server2)
|
||||
|
||||
# Compare with non-Server object
|
||||
self.assertNotEqual(server1a, "not-a-server")
|
||||
self.assertIsNotNone(server1a)
|
||||
|
||||
|
||||
class TestHypervisorWrapper(base.TestCase):
|
||||
"""Test suite for the Hypervisor dataclass."""
|
||||
|
||||
@staticmethod
|
||||
def create_nova_hypervisor(hypervisor_id, hostname, **kwargs):
|
||||
"""Create a real novaclient Hypervisor object.
|
||||
|
||||
:param hypervisor_id: hypervisor UUID
|
||||
:param hostname: hypervisor hostname
|
||||
:param kwargs: additional hypervisor attributes
|
||||
:returns: novaclient.v2.hypervisors.Hypervisor object
|
||||
"""
|
||||
hypervisor_info = {
|
||||
'id': hypervisor_id,
|
||||
'hypervisor_hostname': hostname,
|
||||
'hypervisor_type': kwargs.pop('hypervisor_type', 'QEMU'),
|
||||
'state': kwargs.pop('state', 'up'),
|
||||
'status': kwargs.pop('status', 'enabled'),
|
||||
'vcpus': kwargs.pop('vcpus', 16),
|
||||
'vcpus_used': kwargs.pop('vcpus_used', 4),
|
||||
'memory_mb': kwargs.pop('memory_mb', 32768),
|
||||
'memory_mb_used': kwargs.pop('memory_mb_used', 8192),
|
||||
'local_gb': kwargs.pop('local_gb', 500),
|
||||
'local_gb_used': kwargs.pop('local_gb_used', 100),
|
||||
'service': kwargs.pop('service', {'host': hostname, 'id': 1}),
|
||||
'servers': kwargs.pop('servers', None),
|
||||
}
|
||||
hypervisor_info.update(kwargs)
|
||||
return hypervisors.Hypervisor(
|
||||
hypervisors.HypervisorManager, info=hypervisor_info)
|
||||
|
||||
def test_hypervisor_basic_properties(self):
|
||||
"""Test basic Hypervisor dataclass properties."""
|
||||
hypervisor_id = utils.generate_uuid()
|
||||
hostname = 'compute-node-1'
|
||||
nova_hypervisor = self.create_nova_hypervisor(
|
||||
hypervisor_id,
|
||||
hostname,
|
||||
hypervisor_type='QEMU',
|
||||
state='up',
|
||||
status='enabled',
|
||||
vcpus=32,
|
||||
vcpus_used=8,
|
||||
memory_mb=65536,
|
||||
memory_mb_used=16384,
|
||||
local_gb=1000,
|
||||
local_gb_used=250
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Hypervisor.from_novaclient(nova_hypervisor)
|
||||
|
||||
self.assertEqual(hypervisor_id, wrapped.id)
|
||||
self.assertEqual(hostname, wrapped.hypervisor_hostname)
|
||||
self.assertEqual('QEMU', wrapped.hypervisor_type)
|
||||
self.assertEqual('up', wrapped.state)
|
||||
self.assertEqual('enabled', wrapped.status)
|
||||
self.assertEqual(32, wrapped.vcpus)
|
||||
self.assertEqual(8, wrapped.vcpus_used)
|
||||
self.assertEqual(65536, wrapped.memory_mb)
|
||||
self.assertEqual(16384, wrapped.memory_mb_used)
|
||||
self.assertEqual(1000, wrapped.local_gb)
|
||||
self.assertEqual(250, wrapped.local_gb_used)
|
||||
|
||||
def test_hypervisor_service_properties(self):
|
||||
"""Test Hypervisor dataclass service properties."""
|
||||
hypervisor_id = utils.generate_uuid()
|
||||
hostname = 'compute-node-1'
|
||||
nova_hypervisor = self.create_nova_hypervisor(
|
||||
hypervisor_id,
|
||||
hostname,
|
||||
service={
|
||||
'host': hostname,
|
||||
'id': 42,
|
||||
'disabled_reason': 'maintenance'
|
||||
}
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Hypervisor.from_novaclient(nova_hypervisor)
|
||||
|
||||
self.assertEqual(hostname, wrapped.service_host)
|
||||
self.assertEqual(42, wrapped.service_id)
|
||||
self.assertEqual('maintenance', wrapped.service_disabled_reason)
|
||||
|
||||
def test_hypervisor_service_not_dict(self):
|
||||
"""Test Hypervisor dataclass when service is not a dict."""
|
||||
hypervisor_id = utils.generate_uuid()
|
||||
hostname = 'compute-node-1'
|
||||
nova_hypervisor = self.create_nova_hypervisor(
|
||||
hypervisor_id,
|
||||
hostname,
|
||||
service='not-a-dict'
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Hypervisor.from_novaclient(nova_hypervisor)
|
||||
|
||||
self.assertIsNone(wrapped.service_host)
|
||||
self.assertIsNone(wrapped.service_id)
|
||||
self.assertIsNone(wrapped.service_disabled_reason)
|
||||
|
||||
def test_hypervisor_servers_property(self):
|
||||
"""Test Hypervisor dataclass servers property."""
|
||||
hypervisor_id = utils.generate_uuid()
|
||||
hostname = 'compute-node-1'
|
||||
|
||||
# Create fake server objects with required attributes
|
||||
server1_id = utils.generate_uuid()
|
||||
server2_id = utils.generate_uuid()
|
||||
server1 = {
|
||||
'uuid': server1_id,
|
||||
'name': 'server1',
|
||||
}
|
||||
server2 = {
|
||||
'uuid': server2_id,
|
||||
'name': 'server2'
|
||||
}
|
||||
|
||||
nova_hypervisor = self.create_nova_hypervisor(
|
||||
hypervisor_id,
|
||||
hostname,
|
||||
servers=[server1, server2]
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Hypervisor.from_novaclient(nova_hypervisor)
|
||||
|
||||
# Servers should be wrapped as Server dataclasses
|
||||
result_servers = wrapped.servers
|
||||
self.assertEqual(2, len(result_servers))
|
||||
self.assertEqual(server1_id, result_servers[0]['uuid'])
|
||||
self.assertEqual(server2_id, result_servers[1]['uuid'])
|
||||
|
||||
def test_hypervisor_servers_none(self):
|
||||
"""Test Hypervisor dataclass when servers is None."""
|
||||
hypervisor_id = utils.generate_uuid()
|
||||
hostname = 'compute-node-1'
|
||||
nova_hypervisor = self.create_nova_hypervisor(
|
||||
hypervisor_id,
|
||||
hostname,
|
||||
servers=None
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Hypervisor.from_novaclient(nova_hypervisor)
|
||||
self.assertIsNone(wrapped.servers)
|
||||
|
||||
def test_hypervisor_equality(self):
|
||||
"""Test Hypervisor dataclass equality comparison."""
|
||||
hypervisor_id1 = utils.generate_uuid()
|
||||
hypervisor_id2 = utils.generate_uuid()
|
||||
|
||||
hyp1a = nova_helper.Hypervisor.from_novaclient(
|
||||
self.create_nova_hypervisor(
|
||||
hypervisor_id=hypervisor_id1, hostname='host1'
|
||||
)
|
||||
)
|
||||
hyp1b = nova_helper.Hypervisor.from_novaclient(
|
||||
self.create_nova_hypervisor(
|
||||
hypervisor_id=hypervisor_id1, hostname='host1'
|
||||
)
|
||||
)
|
||||
hyp2 = nova_helper.Hypervisor.from_novaclient(
|
||||
self.create_nova_hypervisor(
|
||||
hypervisor_id=hypervisor_id2, hostname='host2'
|
||||
)
|
||||
)
|
||||
|
||||
# Same ID and attributes should be equal
|
||||
self.assertEqual(hyp1a, hyp1b)
|
||||
|
||||
# Different ID should not be equal
|
||||
self.assertNotEqual(hyp1a, hyp2)
|
||||
|
||||
# Compare with non-Hypervisor object
|
||||
self.assertNotEqual(hyp1a, "not-a-hypervisor")
|
||||
|
||||
|
||||
class TestFlavorWrapper(base.TestCase):
|
||||
"""Test suite for the Flavor dataclass."""
|
||||
|
||||
@staticmethod
|
||||
def create_nova_flavor(flavor_id, name, **kwargs):
|
||||
"""Create a real novaclient Flavor object.
|
||||
|
||||
:param flavor_id: flavor ID
|
||||
:param name: flavor name
|
||||
:param kwargs: additional flavor attributes
|
||||
:returns: novaclient.v2.flavors.Flavor object
|
||||
"""
|
||||
flavor_info = {
|
||||
'id': flavor_id,
|
||||
'name': name,
|
||||
'vcpus': kwargs.pop('vcpus', 2),
|
||||
'ram': kwargs.pop('ram', 2048),
|
||||
'disk': kwargs.pop('disk', 20),
|
||||
'OS-FLV-EXT-DATA:ephemeral': kwargs.pop('ephemeral', 0),
|
||||
'swap': kwargs.pop('swap', ''),
|
||||
'os-flavor-access:is_public': kwargs.pop('is_public', True),
|
||||
}
|
||||
flavor_info.update(kwargs)
|
||||
return flavors.Flavor(flavors.FlavorManager, info=flavor_info)
|
||||
|
||||
def test_flavor_basic_properties(self):
|
||||
"""Test basic Flavor dataclass properties."""
|
||||
flavor_id = utils.generate_uuid()
|
||||
nova_flavor = self.create_nova_flavor(
|
||||
flavor_id,
|
||||
'm1.small',
|
||||
vcpus=2,
|
||||
ram=2048,
|
||||
disk=20,
|
||||
ephemeral=10,
|
||||
swap=512,
|
||||
is_public=True
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Flavor.from_novaclient(nova_flavor)
|
||||
|
||||
self.assertEqual(flavor_id, wrapped.id)
|
||||
self.assertEqual('m1.small', wrapped.flavor_name)
|
||||
self.assertEqual(2, wrapped.vcpus)
|
||||
self.assertEqual(2048, wrapped.ram)
|
||||
self.assertEqual(20, wrapped.disk)
|
||||
self.assertEqual(10, wrapped.ephemeral)
|
||||
self.assertEqual(512, wrapped.swap)
|
||||
self.assertTrue(wrapped.is_public)
|
||||
|
||||
def test_flavor_empty_swap(self):
|
||||
"""Test Flavor dataclass with empty swap string."""
|
||||
flavor_id = utils.generate_uuid()
|
||||
nova_flavor = self.create_nova_flavor(
|
||||
flavor_id,
|
||||
'm1.noswap',
|
||||
swap=''
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Flavor.from_novaclient(nova_flavor)
|
||||
self.assertEqual(0, wrapped.swap)
|
||||
|
||||
def test_flavor_private(self):
|
||||
"""Test Flavor dataclass with private flavor."""
|
||||
flavor_id = utils.generate_uuid()
|
||||
nova_flavor = self.create_nova_flavor(
|
||||
flavor_id,
|
||||
'm1.private',
|
||||
is_public=False
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Flavor.from_novaclient(nova_flavor)
|
||||
self.assertFalse(wrapped.is_public)
|
||||
|
||||
def test_flavor_with_extra_specs(self):
|
||||
"""Test Flavor dataclass with extra_specs."""
|
||||
flavor_id = utils.generate_uuid()
|
||||
nova_flavor = self.create_nova_flavor(
|
||||
flavor_id,
|
||||
'm1.compute',
|
||||
extra_specs={'hw:cpu_policy': 'dedicated', 'hw:numa_nodes': '2'}
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Flavor.from_novaclient(nova_flavor)
|
||||
|
||||
self.assertEqual(
|
||||
{'hw:cpu_policy': 'dedicated', 'hw:numa_nodes': '2'},
|
||||
wrapped.extra_specs
|
||||
)
|
||||
|
||||
def test_flavor_without_extra_specs(self):
|
||||
"""Test Flavor dataclass without extra_specs."""
|
||||
flavor_id = utils.generate_uuid()
|
||||
nova_flavor = self.create_nova_flavor(
|
||||
flavor_id,
|
||||
'm1.basic'
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Flavor.from_novaclient(nova_flavor)
|
||||
self.assertEqual({}, wrapped.extra_specs)
|
||||
|
||||
def test_flavor_equality(self):
|
||||
"""Test Flavor dataclass equality comparison."""
|
||||
flavor_id1 = utils.generate_uuid()
|
||||
flavor_id2 = utils.generate_uuid()
|
||||
|
||||
flavor1a = nova_helper.Flavor.from_novaclient(
|
||||
self.create_nova_flavor(flavor_id1, 'm1.small'))
|
||||
flavor1b = nova_helper.Flavor.from_novaclient(
|
||||
self.create_nova_flavor(flavor_id1, 'm1.small'))
|
||||
flavor2 = nova_helper.Flavor.from_novaclient(
|
||||
self.create_nova_flavor(flavor_id2, 'm1.large'))
|
||||
|
||||
# Same ID and attributes should be equal
|
||||
self.assertEqual(flavor1a, flavor1b)
|
||||
|
||||
# Different ID should not be equal
|
||||
self.assertNotEqual(flavor1a, flavor2)
|
||||
|
||||
# Compare with non-Flavor object
|
||||
self.assertNotEqual(flavor1a, "not-a-flavor")
|
||||
|
||||
|
||||
class TestAggregateWrapper(base.TestCase):
|
||||
"""Test suite for the Aggregate dataclass."""
|
||||
|
||||
@staticmethod
|
||||
def create_nova_aggregate(aggregate_id, name, **kwargs):
|
||||
"""Create a real novaclient Aggregate object.
|
||||
|
||||
:param aggregate_id: aggregate ID
|
||||
:param name: aggregate name
|
||||
:param kwargs: additional aggregate attributes
|
||||
:returns: novaclient.v2.aggregates.Aggregate object
|
||||
"""
|
||||
from novaclient.v2 import aggregates
|
||||
|
||||
aggregate_info = {
|
||||
'id': aggregate_id,
|
||||
'name': name,
|
||||
'availability_zone': kwargs.pop('availability_zone', None),
|
||||
'hosts': kwargs.pop('hosts', []),
|
||||
'metadata': kwargs.pop('metadata', {}),
|
||||
}
|
||||
aggregate_info.update(kwargs)
|
||||
return aggregates.Aggregate(
|
||||
aggregates.AggregateManager, info=aggregate_info)
|
||||
|
||||
def test_aggregate_basic_properties(self):
|
||||
"""Test basic Aggregate dataclass properties."""
|
||||
aggregate_id = utils.generate_uuid()
|
||||
nova_aggregate = self.create_nova_aggregate(
|
||||
aggregate_id,
|
||||
'test-aggregate',
|
||||
availability_zone='az1',
|
||||
hosts=['host1', 'host2', 'host3'],
|
||||
metadata={'ssd': 'true', 'gpu': 'nvidia'}
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Aggregate.from_novaclient(nova_aggregate)
|
||||
|
||||
self.assertEqual(aggregate_id, wrapped.id)
|
||||
self.assertEqual('test-aggregate', wrapped.name)
|
||||
self.assertEqual('az1', wrapped.availability_zone)
|
||||
self.assertEqual(['host1', 'host2', 'host3'], wrapped.hosts)
|
||||
self.assertEqual({'ssd': 'true', 'gpu': 'nvidia'}, wrapped.metadata)
|
||||
|
||||
def test_aggregate_no_az(self):
|
||||
"""Test Aggregate dataclass without availability zone."""
|
||||
aggregate_id = utils.generate_uuid()
|
||||
nova_aggregate = self.create_nova_aggregate(
|
||||
aggregate_id,
|
||||
'test-aggregate',
|
||||
availability_zone=None
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Aggregate.from_novaclient(nova_aggregate)
|
||||
self.assertIsNone(wrapped.availability_zone)
|
||||
|
||||
def test_aggregate_equality(self):
|
||||
"""Test Aggregate dataclass equality comparison."""
|
||||
aggregate_id1 = utils.generate_uuid()
|
||||
aggregate_id2 = utils.generate_uuid()
|
||||
|
||||
agg1a = nova_helper.Aggregate.from_novaclient(
|
||||
self.create_nova_aggregate(aggregate_id1, 'agg1'))
|
||||
agg1b = nova_helper.Aggregate.from_novaclient(
|
||||
self.create_nova_aggregate(aggregate_id1, 'agg1'))
|
||||
agg2 = nova_helper.Aggregate.from_novaclient(
|
||||
self.create_nova_aggregate(aggregate_id2, 'agg2'))
|
||||
|
||||
# Same ID and attributes should be equal
|
||||
self.assertEqual(agg1a, agg1b)
|
||||
|
||||
# Different ID should not be equal
|
||||
self.assertNotEqual(agg1a, agg2)
|
||||
|
||||
# Compare with non-Aggregate object
|
||||
self.assertNotEqual(agg1a, "not-an-aggregate")
|
||||
|
||||
|
||||
class TestServiceWrapper(base.TestCase):
|
||||
"""Test suite for the Service dataclass."""
|
||||
|
||||
@staticmethod
|
||||
def create_nova_service(service_id, **kwargs):
|
||||
"""Create a real novaclient Service object.
|
||||
|
||||
:param service_id: service ID
|
||||
:param kwargs: additional service attributes
|
||||
:returns: novaclient.v2.services.Service object
|
||||
"""
|
||||
from novaclient.v2 import services
|
||||
|
||||
service_info = {
|
||||
'id': service_id,
|
||||
'binary': kwargs.pop('binary', 'nova-compute'),
|
||||
'host': kwargs.pop('host', 'compute-1'),
|
||||
'zone': kwargs.pop('zone', 'nova'),
|
||||
'status': kwargs.pop('status', 'enabled'),
|
||||
'state': kwargs.pop('state', 'up'),
|
||||
'updated_at': kwargs.pop('updated_at', '2026-01-09T12:00:00Z'),
|
||||
'disabled_reason': kwargs.pop('disabled_reason', None),
|
||||
}
|
||||
service_info.update(kwargs)
|
||||
return services.Service(services.ServiceManager, info=service_info)
|
||||
|
||||
def test_service_basic_properties(self):
|
||||
"""Test basic Service dataclass properties."""
|
||||
service_id = utils.generate_uuid()
|
||||
nova_service = self.create_nova_service(
|
||||
service_id,
|
||||
binary='nova-compute',
|
||||
host='compute-node-1',
|
||||
zone='az1',
|
||||
status='enabled',
|
||||
state='up',
|
||||
updated_at='2026-01-09T12:00:00Z',
|
||||
disabled_reason=None
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Service.from_novaclient(nova_service)
|
||||
|
||||
self.assertEqual(service_id, wrapped.id)
|
||||
self.assertEqual('nova-compute', wrapped.binary)
|
||||
self.assertEqual('compute-node-1', wrapped.host)
|
||||
self.assertEqual('az1', wrapped.zone)
|
||||
self.assertEqual('enabled', wrapped.status)
|
||||
self.assertEqual('up', wrapped.state)
|
||||
self.assertEqual('2026-01-09T12:00:00Z', wrapped.updated_at)
|
||||
self.assertIsNone(wrapped.disabled_reason)
|
||||
|
||||
def test_service_disabled(self):
|
||||
"""Test Service dataclass with disabled service."""
|
||||
service_id = utils.generate_uuid()
|
||||
nova_service = self.create_nova_service(
|
||||
service_id,
|
||||
status='disabled',
|
||||
state='down',
|
||||
disabled_reason='maintenance'
|
||||
)
|
||||
|
||||
wrapped = nova_helper.Service.from_novaclient(nova_service)
|
||||
|
||||
self.assertEqual('disabled', wrapped.status)
|
||||
self.assertEqual('down', wrapped.state)
|
||||
self.assertEqual('maintenance', wrapped.disabled_reason)
|
||||
|
||||
def test_service_equality(self):
|
||||
"""Test Service dataclass equality comparison."""
|
||||
service_id1 = utils.generate_uuid()
|
||||
service_id2 = utils.generate_uuid()
|
||||
|
||||
svc1a = nova_helper.Service.from_novaclient(
|
||||
self.create_nova_service(service_id1))
|
||||
svc1b = nova_helper.Service.from_novaclient(
|
||||
self.create_nova_service(service_id1))
|
||||
svc2 = nova_helper.Service.from_novaclient(
|
||||
self.create_nova_service(service_id2))
|
||||
|
||||
# Same ID and attributes should be equal
|
||||
self.assertEqual(svc1a, svc1b)
|
||||
|
||||
# Different ID should not be equal
|
||||
self.assertNotEqual(svc1a, svc2)
|
||||
|
||||
# Compare with non-Service object
|
||||
self.assertNotEqual(svc1a, "not-a-service")
|
||||
|
||||
|
||||
class TestServerMigrationWrapper(base.TestCase):
|
||||
"""Test suite for the ServerMigration dataclass."""
|
||||
|
||||
@staticmethod
|
||||
def create_nova_migration(migration_id, **kwargs):
|
||||
"""Create a real novaclient ServerMigration object.
|
||||
|
||||
:param migration_id: migration ID
|
||||
:param kwargs: additional migration attributes
|
||||
:returns: novaclient.v2.server_migrations.ServerMigration object
|
||||
"""
|
||||
from novaclient.v2 import server_migrations
|
||||
|
||||
migration_info = {
|
||||
'id': migration_id,
|
||||
}
|
||||
migration_info.update(kwargs)
|
||||
return server_migrations.ServerMigration(
|
||||
server_migrations.ServerMigrationsManager, info=migration_info)
|
||||
|
||||
def test_migration_basic_properties(self):
|
||||
"""Test basic ServerMigration dataclass properties."""
|
||||
migration_id = utils.generate_uuid()
|
||||
nova_migration = self.create_nova_migration(migration_id)
|
||||
|
||||
wrapped = nova_helper.ServerMigration.from_novaclient(nova_migration)
|
||||
self.assertEqual(migration_id, wrapped.id)
|
||||
|
||||
def test_migration_equality(self):
|
||||
"""Test ServerMigration dataclass equality comparison."""
|
||||
migration_id1 = utils.generate_uuid()
|
||||
migration_id2 = utils.generate_uuid()
|
||||
|
||||
mig1a = nova_helper.ServerMigration.from_novaclient(
|
||||
self.create_nova_migration(migration_id1))
|
||||
mig1b = nova_helper.ServerMigration.from_novaclient(
|
||||
self.create_nova_migration(migration_id1))
|
||||
mig2 = nova_helper.ServerMigration.from_novaclient(
|
||||
self.create_nova_migration(migration_id2))
|
||||
|
||||
# Same ID and attributes should be equal
|
||||
self.assertEqual(mig1a, mig1b)
|
||||
|
||||
# Different ID should not be equal
|
||||
self.assertNotEqual(mig1a, mig2)
|
||||
|
||||
# Compare with non-ServerMigration object
|
||||
self.assertNotEqual(mig1a, "not-a-migration")
|
||||
|
||||
Reference in New Issue
Block a user