OpenShift introduction

server/src/uds/services/OpenShift/provider.py

@@ -126,10 +126,11 @@ def test(
     def sanitized_name(self, name: str) -> str:
         """
         Sanitizes the VM name to comply with RFC 1123:
-        - Lowercase
-        - Alphanumeric, '-', '.'
-        - Starts/ends with alphanumeric
-        - Max length 63 chars
+        - Converts to lowercase
+        - Replaces any character not in [a-z0-9.-] with '-'
+        - Removes leading/trailing non-alphanumeric characters
+        - Trims leading/trailing '-' or '.'
+        - Limits length to 63 characters
         """
-        name = re.sub(r'^[^a-z0-9]+|[^a-z0-9.-]|-{2,}|[^a-z0-9]+$', '-', name.lower())
+        name = re.sub(r'^[^a-z0-9]+|[^a-z0-9]+$|[^a-z0-9.-]', '-', name.lower()).strip('-.')
         return name[:63]

server/tests/services/openshift/fixtures.py

@@ -0,0 +1,348 @@
+# -*- coding: utf-8 -*-
+"""
+Test fixtures for OpenShift service tests.
+Provides reusable functions and mock objects for unit testing OpenShift provider, service, deployment, publication, and user service logic.
+All functions are designed to be used across multiple test modules for consistency and maintainability.
+"""
+
+#
+# Copyright (c) 2024 Virtual Cable S.L.U.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without modification,
+# are permitted provided that the following conditions are met:
+#
+#    * Redistributions of source code must retain the above copyright notice,
+#      this list of conditions and the following disclaimer.
+#    * Redistributions in binary form must reproduce the above copyright notice,
+#      this list of conditions and the following disclaimer in the documentation
+#      and/or other materials provided with the distribution.
+#    * Neither the name of Virtual Cable S.L.U. nor the names of its contributors
+#      may be used to endorse or promote products derived from this software
+#      without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+"""
+Author: Adolfo Gómez, dkmaster at dkmon dot com
+"""
+import contextlib
+import copy
+import functools
+import random
+import typing
+
+from unittest import mock
+import uuid
+
+
+from uds.core import environment
+from uds.core.ui.user_interface import gui
+from uds.models.user import User
+
+from uds.services.OpenShift import service, service_fixed, provider, publication, deployment, deployment_fixed
+from uds.services.OpenShift.openshift import types as openshift_types, exceptions as openshift_exceptions
+
+DEF_VMS: list[openshift_types.VM] = [
+    openshift_types.VM(
+        name=f'vm-{i}',
+        namespace='default',
+        uid=f'uid-{i}',
+        status=openshift_types.VMStatus.STOPPED if i % 2 == 0 else openshift_types.VMStatus.RUNNING,
+        volume_template=openshift_types.VolumeTemplate(name=f'volume-{i}', storage='10Gi'),
+        disks=[openshift_types.DeviceDisk(name=f'disk-{i}', boot_order=1)],
+        volumes=[openshift_types.Volume(name=f'volume-{i}', data_volume=f'dv-{i}')],
+    )
+    for i in range(1, 11)
+]
+DEF_VM_INSTANCES: list[openshift_types.VMInstance] = [
+    openshift_types.VMInstance(
+        name=f'vm-{i}',
+        namespace='default',
+        uid=f'uid-instance-{i}',
+        interfaces=[
+            openshift_types.Interface(
+                name='eth0',
+                mac_address=f'00:11:22:33:44:{i:02x}',
+                ip_address=f'192.168.1.{i}',
+            )
+        ],
+        status=openshift_types.VMStatus.STOPPED if i % 2 == 0 else openshift_types.VMStatus.RUNNING,
+        phase=openshift_types.VMStatus.STOPPED if i % 2 == 0 else openshift_types.VMStatus.RUNNING,
+    )
+    for i in range(1, 11)
+]
+
+# clone values to avoid modifying the original ones
+VMS: list[openshift_types.VM] = copy.deepcopy(DEF_VMS)
+VM_INSTANCES: list[openshift_types.VMInstance] = copy.deepcopy(DEF_VM_INSTANCES)
+
+
+def clear() -> None:
+    """
+    Reset all VM and VMInstance values to their default state.
+    Use this before each test to ensure a clean environment.
+    """
+    VMS[:] = copy.deepcopy(DEF_VMS)
+    VM_INSTANCES[:] = copy.deepcopy(DEF_VM_INSTANCES)
+
+
+def replace_vm_info(vm_name: str, **kwargs: typing.Any) -> None:
+    """
+    Update attributes of a VM in VMS by name.
+    Raises OpenshiftNotFoundError if VM is not found.
+    """
+    try:
+        vm = next(vm for vm in VMS if vm.name == vm_name)
+        for k, v in kwargs.items():
+            setattr(vm, k, v)
+    except Exception:
+        raise openshift_exceptions.OpenshiftNotFoundError(f'VM {vm_name} not found')
+
+
+def replacer_vm_info(**kwargs: typing.Any) -> typing.Callable[..., None]:
+    """
+    Returns a partial function to update VM info with preset kwargs.
+    Useful for patching or repeated updates in tests.
+    """
+    return functools.partial(replace_vm_info, **kwargs)
+
+
+T = typing.TypeVar('T')
+
+
+def returner(value: T, *args: typing.Any, **kwargs: typing.Any) -> typing.Callable[..., T]:
+    """
+    Returns a function that always returns the given value.
+    Useful for mocking return values in tests.
+    """
+    def inner(*args: typing.Any, **kwargs: typing.Any) -> T:
+        return value
+
+    return inner
+
+
+# Provider values
+PROVIDER_VALUES_DICT: gui.ValuesDictType = {
+    'cluster_url': 'https://oauth-openshift.apps-crc.testing',
+    'api_url': 'https://api.crc.testing:6443',
+    'username': 'kubeadmin',
+    'password': 'test-password',
+    'namespace': 'default',
+    'verify_ssl': False,
+    'concurrent_creation_limit': 1,
+    'concurrent_removal_limit': 1,
+    'timeout': 10,
+}
+
+# Service values
+SERVICE_VALUES_DICT: gui.ValuesDictType = {
+    'template': VMS[0].name,
+    'basename': 'base',
+    'lenname': 4,
+    'publication_timeout': 120,
+    'prov_uuid': '',
+}
+
+# Service fixed values
+SERVICE_FIXED_VALUES_DICT: gui.ValuesDictType = {
+    'token': '',
+    'machines': [VMS[2].name, VMS[3].name, VMS[4].name],
+    'on_logout': 'no',
+    'randomize': False,
+    'maintain_on_error': False,
+    'prov_uuid': '',
+}
+
+
+def create_client_mock() -> mock.Mock:
+    """
+    Create a MagicMock for OpenshiftClient with default behaviors and side effects.
+    Used to simulate API responses in provider/service tests.
+    """
+    client = mock.MagicMock()
+
+    # Prepare deep copies of default data
+    client.test.return_value = True
+    client.list_vms.return_value = copy.deepcopy(DEF_VMS)
+    client.start_vm_instance.return_value = True
+    client.stop_vm_instance.return_value = True
+    client.delete_vm_instance.return_value = True
+    client.get_datavolume_phase.return_value = "Succeeded"
+    client.get_vm_pvc_or_dv_name.return_value = ("test-pvc", "pvc")
+    client.get_pvc_size.return_value = "10Gi"
+    client.create_vm_from_pvc.return_value = True
+    client.wait_for_datavolume_clone_progress.return_value = True
+
+    def get_vm_info_side_effect(vm_name: str, **kwargs: typing.Any) -> openshift_types.VM | None:
+        for vm in VMS:
+            if vm.name == vm_name:
+                return vm
+        return None
+
+    def get_vm_instance_info_side_effect(vm_name: str, **kwargs: typing.Any) -> openshift_types.VMInstance | None:
+        for inst in VM_INSTANCES:
+            if inst.name == vm_name:
+                return inst
+        return None
+
+    client.get_vm_info.side_effect = get_vm_info_side_effect
+    client.get_vm_instance_info.side_effect = get_vm_instance_info_side_effect
+
+    return client
+
+
+@contextlib.contextmanager
+def patched_provider(**kwargs: typing.Any) -> typing.Generator[provider.OpenshiftProvider, None, None]:
+    """
+    Context manager that yields a provider with a patched OpenshiftClient mock.
+    Use this to ensure all API calls are intercepted and controlled in tests.
+    """
+    client = create_client_mock()
+    prov = create_provider(**kwargs)
+    prov._cached_api = client
+    yield prov
+
+
+def create_provider(**kwargs: typing.Any) -> provider.OpenshiftProvider:
+    """
+    Create an OpenshiftProvider instance with default or overridden values.
+    Used for provider-level tests and as a dependency for other fixtures.
+    """
+    values = PROVIDER_VALUES_DICT.copy()
+    values.update(kwargs)
+
+    uuid_ = str(uuid.uuid4())
+    return provider.OpenshiftProvider(
+        environment=environment.Environment.private_environment(uuid_), values=values, uuid=uuid_
+    )
+
+
+def create_service(
+    provider: typing.Optional[provider.OpenshiftProvider] = None, **kwargs: typing.Any
+) -> service.OpenshiftService:
+    """
+    Create an OpenshiftService instance (dynamic service).
+    Used for service-level tests and as a dependency for user services and publications.
+    """
+    uuid_ = str(uuid.uuid4())
+    values = SERVICE_VALUES_DICT.copy()
+    values.update(kwargs)
+    srvc = service.OpenshiftService(
+        environment=environment.Environment.private_environment(uuid_),
+        provider=provider or create_provider(),
+        values=values,
+        uuid=uuid_,
+    )
+    return srvc
+
+
+def create_service_fixed(
+    provider: typing.Optional[provider.OpenshiftProvider] = None, **kwargs: typing.Any
+) -> service_fixed.OpenshiftServiceFixed:
+    """
+    Create an OpenshiftServiceFixed instance (fixed service).
+    Used for fixed service tests and as a dependency for fixed user services.
+    """
+    uuid_ = str(uuid.uuid4())
+    values = SERVICE_FIXED_VALUES_DICT.copy()
+    values.update(kwargs)
+    return service_fixed.OpenshiftServiceFixed(
+        environment=environment.Environment.private_environment(uuid_),
+        provider=provider or create_provider(),
+        values=values,
+        uuid=uuid_,
+    )
+
+
+def create_publication(
+    service: typing.Optional[service.OpenshiftService] = None,
+    **kwargs: typing.Any,
+) -> publication.OpenshiftTemplatePublication:
+    """
+    Create an OpenshiftTemplatePublication instance.
+    Used for publication-level tests and as a dependency for user services.
+    """
+    uuid_ = str(uuid.uuid4())
+    pub = publication.OpenshiftTemplatePublication(
+        environment=environment.Environment.private_environment(uuid_),
+        service=service or create_service(**kwargs),
+        revision=1,
+        servicepool_name='servicepool_name',
+        uuid=uuid_,
+    )
+    pub._name = f"pub-{random.randint(1000, 9999)}"
+    return pub
+
+
+def create_userservice(
+    service: typing.Optional[service.OpenshiftService] = None,
+    publication: typing.Optional[publication.OpenshiftTemplatePublication] = None,
+) -> deployment.OpenshiftUserService:
+    """
+    Create an OpenshiftUserService instance (dynamic user service).
+    Used for user service tests that require a publication and service.
+    """
+    uuid_ = str(uuid.uuid4())
+    return deployment.OpenshiftUserService(
+        environment=environment.Environment.private_environment(uuid_),
+        service=service or create_service(),
+        publication=publication or create_publication(),
+        uuid=uuid_,
+    )
+
+
+def create_userservice_fixed(
+    service: typing.Optional[service_fixed.OpenshiftServiceFixed] = None,
+) -> deployment_fixed.OpenshiftUserServiceFixed:
+    """
+    Create an OpenshiftUserServiceFixed instance (fixed user service).
+    Used for tests of fixed user service logic and lifecycle.
+    """
+    uuid_ = str(uuid.uuid4().hex)
+    return deployment_fixed.OpenshiftUserServiceFixed(
+        environment=environment.Environment.private_environment(uuid_),
+        service=service or create_service_fixed(),
+        publication=None,
+        uuid=uuid_,
+    )
+
+
+def create_user(
+    name: str = "testuser",
+    real_name: str = "Test User",
+    is_admin: bool = False,
+    state: str = 'A',
+    password: str = 'password',
+    mfa_data: str = '',
+    staff_member: bool = False,
+    last_access: typing.Optional[str] = None,
+    parent: typing.Optional[User] = None,
+    created: typing.Optional[str] = None,
+    comments: str = '',
+) -> User:
+    """
+    Create a mock User instance for testing.
+    All fields can be customized for specific test scenarios.
+    """
+    user = mock.Mock(spec=User)
+    user.name = name
+    user.real_name = real_name
+    user.is_admin = is_admin
+    user.state = state
+    user.password = password
+    user.mfa_data = mfa_data
+    user.staff_member = staff_member
+    user.last_access = last_access
+    user.parent = parent
+    user.created = created
+    user.comments = comments
+    return user