add test for validating network actions

Author: saklar13

cloudshell/cp/vcenter/flows/connectivity_flow.py

@@ -18,7 +18,6 @@
     is_set_action,
 )
 
-from cloudshell.cp.vcenter.exceptions import BaseVCenterException
 from cloudshell.cp.vcenter.handlers.dc_handler import DcHandler
 from cloudshell.cp.vcenter.handlers.managed_entity_handler import ManagedEntityNotFound
 from cloudshell.cp.vcenter.handlers.network_handler import (
@@ -65,15 +64,6 @@
 switch_lock = LockHandler()
 
 
-class DvSwitchNameEmpty(BaseVCenterException):
-    def __init__(self):
-        msg = (
-            "For connectivity actions you have to specify default DvSwitch name in the "
-            "resource or in every VLAN service"
-        )
-        super().__init__(msg)
-
-
 @define(slots=False)
 class VCenterConnectivityFlow(AbcCloudProviderConnectivityFlow):
     _si: SiHandler
@@ -105,11 +95,9 @@ def _holding_network(self) -> NetworkHandler | DVPortGroupHandler:
     def validate_actions(
         self, actions: Collection[VcenterConnectivityActionModel]
     ) -> None:
-        for net_settings in map(self._get_network_settings, actions):
-            without_switch = not net_settings.switch_name
-            new_network = not net_settings.existed
-            if without_switch and new_network:
-                raise DvSwitchNameEmpty
+        # if switch name not specified in VLAN service or in resource config
+        # converter would raise an exception
+        _ = [self._get_network_settings(action) for action in actions]
 
     def pre_connectivity(
         self,

cloudshell/cp/vcenter/utils/connectivity_helpers.py

@@ -40,6 +40,15 @@
 PORT_GROUP_NAME_PATTERN = re.compile(rf"{QS_NAME_PREFIX}_.+_VLAN")
 
 
+class DvSwitchNameEmpty(BaseVCenterException):
+    def __init__(self):
+        msg = (
+            "For connectivity actions you have to specify default DvSwitch name in the "
+            "resource or in every VLAN service"
+        )
+        super().__init__(msg)
+
+
 @define
 class PgCanNotBeRemoved(BaseVCenterException):
     name: str
@@ -222,6 +231,8 @@ def convert(
         if name := (old_name := get_existed_port_group_name(action)):
             existed = True
         else:
+            if not switch:
+                raise DvSwitchNameEmpty
             existed = False
             old_name = generate_port_group_name(switch, vlan_id, port_mode)
             name = generate_port_group_name_v2(

tests/conftest.py

@@ -2,9 +2,11 @@
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
+from pyVmomi import vim, vmodl
 
 from cloudshell.cp.core.cancellation_manager import CancellationContextManager
 from cloudshell.cp.core.request_actions import GetVMDetailsRequestActions
+from cloudshell.cp.core.reservation_info import ReservationInfo
 from cloudshell.shell.core.driver_context import (
     AppContext,
     ConnectivityContext,
@@ -18,10 +20,12 @@
 from cloudshell.cp.vcenter.constants import SHELL_NAME, STATIC_SHELL_NAME
 from cloudshell.cp.vcenter.handlers.cluster_handler import ClusterHandler
 from cloudshell.cp.vcenter.handlers.dc_handler import DcHandler
+from cloudshell.cp.vcenter.handlers.managed_entity_handler import ManagedEntityHandler
 from cloudshell.cp.vcenter.handlers.si_handler import SiHandler
 from cloudshell.cp.vcenter.handlers.vm_handler import VmHandler
 from cloudshell.cp.vcenter.models.deployed_app import StaticVCenterDeployedApp
 from cloudshell.cp.vcenter.resource_config import VCenterResourceConfig
+from cloudshell.cp.vcenter.utils.network_watcher import NetworkWatcher
 
 
 @pytest.fixture()
@@ -83,6 +87,11 @@ def reservation_context_details() -> ReservationContextDetails:
     )
 
 
+@pytest.fixture
+def reservation_info(reservation_context_details):
+    return ReservationInfo._from_reservation_context(reservation_context_details)
+
+
 @pytest.fixture()
 def resource_command_context(
     connectivity_context, resource_context_details, reservation_context_details
@@ -122,7 +131,7 @@ def resource_conf(resource_command_context, cs_api) -> VCenterResourceConfig:
     promiscuous_mode = "true"
     forged_transmits = "true"
     mac_address_changes = "false"
-    enable_tags = "true"
+    enable_tags = "false"
 
     a_name = VCenterResourceConfig.ATTR_NAMES
     get_full_a_name = lambda n: f"{SHELL_NAME}.{n}"  # noqa: E731
@@ -151,7 +160,14 @@ def resource_conf(resource_command_context, cs_api) -> VCenterResourceConfig:
             get_full_a_name(a_name.enable_tags): enable_tags,
         }
     )
-    conf = VCenterResourceConfig.from_context(resource_command_context, cs_api)
+
+    class Cfg(VCenterResourceConfig):
+        __dict__ = {}
+
+        def __setattr__(self, key, value):
+            object.__setattr__(self, key, value)
+
+    conf = Cfg.from_context(resource_command_context, cs_api)
 
     return conf
 
@@ -311,3 +327,42 @@ def static_deployed_app(cs_api) -> StaticVCenterDeployedApp:
     GetVMDetailsRequestActions.register_deployment_path(StaticVCenterDeployedApp)
     actions = GetVMDetailsRequestActions.from_request(requests, cs_api)
     return actions.deployed_apps[0]
+
+
+@pytest.fixture()
+def object_spec(monkeypatch):
+    m = Mock()
+    monkeypatch.setattr(vmodl.query.PropertyCollector, "ObjectSpec", m)
+    return m
+
+
+@pytest.fixture()
+def filter_spec(monkeypatch):
+    m = Mock()
+    monkeypatch.setattr(vmodl.query.PropertyCollector, "FilterSpec", m)
+    return m
+
+
+@pytest.fixture()
+def property_collector(si):
+    m = Mock()
+    si.get_vc_obj().content.propertyCollector.CreatePropertyCollector.return_value = m
+    # no updates
+    m.WaitForUpdatesEx.side_effect = [None]
+    return m
+
+
+@pytest.fixture()
+def container(si):
+    vc_container = Mock(spec=vim.Datacenter)
+    vc_container.name = "Datacenter"
+    return ManagedEntityHandler(vc_container, si)
+
+
+@pytest.fixture()
+def network_watcher(si, container, object_spec, filter_spec, property_collector):
+    # add ability to modify class attributes
+    class N(NetworkWatcher):
+        __dict__ = {}
+
+    return N(si, container)

tests/cp/vcenter/flows/test_connectivity_flow.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from cloudshell.shell.flows.connectivity.parse_request_service import (
+    ParseConnectivityRequestService,
+)
+
+from cloudshell.cp.vcenter.flows.connectivity_flow import VCenterConnectivityFlow
+from cloudshell.cp.vcenter.models.connectivity_action_model import (
+    VcenterConnectivityActionModel,
+)
+from cloudshell.cp.vcenter.utils.connectivity_helpers import DvSwitchNameEmpty
+
+logger = logging.getLogger(__name__)
+
+
+ACTION_DICT = {
+    "connectionId": "96582265-2728-43aa-bc97-cefb2457ca44",
+    "connectionParams": {
+        "vlanId": "11",
+        "mode": "Access",
+        "vlanServiceAttributes": [
+            {
+                "attributeName": "QnQ",
+                "attributeValue": "False",
+                "type": "vlanServiceAttribute",
+            },
+            {
+                "attributeName": "CTag",
+                "attributeValue": "",
+                "type": "vlanServiceAttribute",
+            },
+            {
+                "attributeName": "VLAN ID",
+                "attributeValue": "11",
+                "type": "vlanServiceAttribute",
+            },
+            {
+                "attributeName": "Virtual Network",
+                "attributeValue": "",
+                "type": "vlanServiceAttribute",
+            },
+        ],
+        "type": "setVlanParameter",
+    },
+    "connectorAttributes": [
+        {
+            "attributeName": "Interface",
+            "attributeValue": "mac address",
+            "type": "connectorAttribute",
+        },
+    ],
+    "actionTarget": {
+        "fullName": "centos",
+        "fullAddress": "full address",
+        "type": "actionTarget",
+    },
+    "customActionAttributes": [
+        {
+            "attributeName": "VM_UUID",
+            "attributeValue": "vm_uid",
+            "type": "customAttribute",
+        },
+        {
+            "attributeName": "Vnic Name",
+            "attributeValue": "vnic",
+            "type": "customAttribute",
+        },
+    ],
+    "actionId": "96582265-2728-43aa-bc97-cefb2457ca44_0900c4b5-0f90-42e3-b495",
+    "type": "setVlan",
+}
+
+
+@pytest.fixture
+def parse_connectivity_service():
+    return ParseConnectivityRequestService(
+        is_vlan_range_supported=True, is_multi_vlan_supported=True
+    )
+
+
+@pytest.fixture
+def flow(
+    si, parse_connectivity_service, resource_conf, reservation_info, dc, network_watcher
+):
+    return VCenterConnectivityFlow(
+        parse_connectivity_service, si, resource_conf, reservation_info
+    )
+
+
+@pytest.fixture
+def set_action():
+    return VcenterConnectivityActionModel.model_validate(ACTION_DICT)
+
+
+def test_validate_actions(flow, set_action):
+    flow.validate_actions([set_action])
+
+
+def test_validate_actions_without_switch(flow, set_action, resource_conf):
+    resource_conf.default_dv_switch = None
+    with pytest.raises(DvSwitchNameEmpty):
+        flow.validate_actions([set_action])
+
+
+def test_validate_actions_switch_in_vlan_service(flow, set_action, resource_conf):
+    resource_conf.default_dv_switch = None
+    set_action.connection_params.vlan_service_attrs.switch_name = "switch"
+    flow.validate_actions([set_action])
+
+
+def test_validate_actions_switch_is_empty_but_we_use_user_created_network(
+    flow, set_action, resource_conf
+):
+    resource_conf.default_dv_switch = None
+    set_action.connection_params.vlan_service_attrs.existing_network = "network"
+    flow.validate_actions([set_action])

tests/cp/vcenter/flows/test_save_restore.py

@@ -99,7 +99,7 @@ def test_save(flow, vm, dc_handler, resource_conf):
     vm_uuid = vm.uuid
     action_id = "action id"
     action = _get_save_action(vm_uuid, "Inherited", action_id)
-    resource_conf.__dict__["behavior_during_save"] = "Power Off"
+    resource_conf.behavior_during_save = "Power Off"
     cloned_vm_name = f"Clone of {vm.name}"
     cloned_vm = Mock(name=cloned_vm_name, path=f"folder/{cloned_vm_name}", uuid="uuid")
     cloned_vm.name = cloned_vm_name

tests/cp/vcenter/utils/test_network_watcher.py

@@ -5,55 +5,16 @@
 from unittest.mock import Mock
 
 import pytest
-from pyVmomi import vim, vmodl
+from pyVmomi import vim
 
-from cloudshell.cp.vcenter.handlers.managed_entity_handler import ManagedEntityHandler
 from cloudshell.cp.vcenter.handlers.network_handler import (
     NetworkHandler,
     NetworkNotFound,
 )
-from cloudshell.cp.vcenter.utils.network_watcher import NetworkWatcher
 
 logger = logging.getLogger(__name__)
 
 
-@pytest.fixture()
-def object_spec(monkeypatch):
-    m = Mock()
-    monkeypatch.setattr(vmodl.query.PropertyCollector, "ObjectSpec", m)
-    yield m
-
-
-@pytest.fixture()
-def filter_spec(monkeypatch):
-    m = Mock()
-    monkeypatch.setattr(vmodl.query.PropertyCollector, "FilterSpec", m)
-    yield m
-
-
-@pytest.fixture()
-def property_collector(si):
-    m = Mock()
-    si.get_vc_obj().content.propertyCollector.CreatePropertyCollector.return_value = m
-    yield m
-
-
-@pytest.fixture()
-def container(si):
-    vc_container = Mock(spec=vim.Datacenter)
-    vc_container.name = "Datacenter"
-    return ManagedEntityHandler(vc_container, si)
-
-
-@pytest.fixture()
-def network_watcher(si, container, object_spec, filter_spec, property_collector):
-    # add ability to modify class attributes
-    class N(NetworkWatcher):
-        __dict__ = {}
-
-    return N(si, container)
-
-
 @pytest.fixture()
 def update_network_do_nothing(monkeypatch, network_watcher):
     m = Mock()