"""Service onboarding step module."""
+
import time
from typing import Any, Dict, Iterator
from urllib.parse import urlencode
from onapsdk.aai.service_design_and_creation import Model
from onapsdk.configuration import settings
from onapsdk.exceptions import InvalidResponse, ResourceNotFound
-from onapsdk.sdc2.component_instance import (ComponentInstance,
- ComponentInstanceInput)
+from onapsdk.sdc2.component_instance import ComponentInstance, ComponentInstanceInput
from onapsdk.sdc2.pnf import Pnf
from onapsdk.sdc2.sdc_category import ServiceCategory
from onapsdk.sdc2.sdc_resource import LifecycleOperation, LifecycleState
return self.model_yaml_template
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r", encoding="utf-8") as yaml_template:
+ with open(
+ settings.SERVICE_YAML_TEMPLATE, "r", encoding="utf-8"
+ ) as yaml_template:
self._yaml_template: dict = load(yaml_template, SafeLoader)
return self._yaml_template
return self.parent.yaml_template
"""
if self.is_root:
if not self._model_yaml_template:
- with open(settings.MODEL_YAML_TEMPLATE, "r",
- encoding="utf-8") as model_yaml_template:
- self._model_yaml_template: dict = load(model_yaml_template, SafeLoader)
+ with open(
+ settings.MODEL_YAML_TEMPLATE, "r", encoding="utf-8"
+ ) as model_yaml_template:
+ self._model_yaml_template: dict = load(
+ model_yaml_template, SafeLoader
+ )
return self._model_yaml_template
return self.parent.model_yaml_template
super().execute()
if "instantiation_type" in self.yaml_template[self.service_name]:
instantiation_type: ServiceInstantiationType = ServiceInstantiationType(
- self.yaml_template[self.service_name]["instantiation_type"])
+ self.yaml_template[self.service_name]["instantiation_type"]
+ )
else:
- instantiation_type: ServiceInstantiationType = ServiceInstantiationType.A_LA_CARTE
+ instantiation_type: ServiceInstantiationType = (
+ ServiceInstantiationType.A_LA_CARTE
+ )
with tracer.start_as_current_span(
- "sdc.service.get_or_create",
- attributes={"service.name": self.service_name}
+ "sdc.service.get_or_create", attributes={"service.name": self.service_name}
) as sdc_span:
try:
service: Service = Service.get_by_name(name=self.service_name)
except ResourceNotFound:
sdc_span.set_attribute("service.existed", False)
self._logger.info("before service create")
- service = Service.create(name=self.service_name,
- instantiation_type=instantiation_type,
- category=category)
+ service = Service.create(
+ name=self.service_name,
+ instantiation_type=instantiation_type,
+ category=category,
+ )
self._logger.info("after service create")
- sdc_span.set_attribute("service.instantiation_type", instantiation_type.value)
+ sdc_span.set_attribute(
+ "service.instantiation_type", instantiation_type.value
+ )
self.declare_resources(service)
self.assign_properties(service)
"sdc.service.certify",
attributes={
"service.name": self.service_name,
- "service.lifecycle_state": service.lifecycle_state.value
- }
+ "service.lifecycle_state": service.lifecycle_state.value,
+ },
):
service.lifecycle_operation(LifecycleOperation.CERTIFY)
+ service.distribute()
def declare_resources(self, service: Service) -> None:
"""Declare resources.
"""
if "networks" in self.yaml_template[self.service_name]:
for net in self.yaml_template[self.service_name]["networks"]:
- vl: Vl = Vl.get_by_name(name=net['vl_name'])
+ vl: Vl = Vl.get_by_name(name=net["vl_name"])
service.add_resource(vl)
if "vnfs" in self.yaml_template[self.service_name]:
for vnf in self.yaml_template[self.service_name]["vnfs"]:
if "networks" in self.yaml_template[self.service_name]:
for net in self.yaml_template[self.service_name]["networks"]:
if "properties" in net:
- vl_component: ComponentInstance = service.get_component_by_name(net['vl_name'])
+ vl_component: ComponentInstance = service.get_component_by_name(
+ net["vl_name"]
+ )
self.assign_properties_to_component(vl_component, net["properties"])
if "vnfs" in self.yaml_template[self.service_name]:
for vnf in self.yaml_template[self.service_name]["vnfs"]:
if "properties" in vnf:
- vf_component: ComponentInstance = service.get_component_by_name(vnf["vnf_name"])
+ vf_component: ComponentInstance = service.get_component_by_name(
+ vnf["vnf_name"]
+ )
self.assign_properties_to_component(vf_component, vnf["properties"])
if "pnfs" in self.yaml_template[self.service_name]:
for pnf in self.yaml_template[self.service_name]["pnfs"]:
if "properties" in pnf:
- pnf_component: ComponentInstance = \
- service.get_component_by_name(pnf["pnf_name"])
- self.assign_properties_to_component(pnf_component, pnf["properties"])
-
- def assign_properties_to_component(self,
- component: ComponentInstance,
- component_properties: Dict[str, Any]) -> None:
+ pnf_component: ComponentInstance = service.get_component_by_name(
+ pnf["pnf_name"]
+ )
+ self.assign_properties_to_component(
+ pnf_component, pnf["properties"]
+ )
+
+ def assign_properties_to_component(
+ self, component: ComponentInstance, component_properties: Dict[str, Any]
+ ) -> None:
"""Assign properties to component.
Args:
def cleanup(self) -> None:
"""Cleanup service onboard step."""
with tracer.start_as_current_span(
- "sdc.service.delete",
- attributes={"service.name": self.service_name}
+ "sdc.service.delete", attributes={"service.name": self.service_name}
) as cleanup_span:
try:
service: Service = Service.get_by_name(name=self.service_name)
cleanup_span.set_attribute("service.found", True)
- cleanup_span.set_attribute("service.lifecycle_state", service.lifecycle_state.value)
+ cleanup_span.set_attribute(
+ "service.lifecycle_state", service.lifecycle_state.value
+ )
if service.lifecycle_state == LifecycleState.CERTIFIED:
service.archive()
service.delete()
return self.model_yaml_template
if self.is_root:
if not self._yaml_template:
- with open(settings.SERVICE_YAML_TEMPLATE, "r", encoding="utf-8") as yaml_template:
+ with open(
+ settings.SERVICE_YAML_TEMPLATE, "r", encoding="utf-8"
+ ) as yaml_template:
self._yaml_template: dict = load(yaml_template, SafeLoader)
return self._yaml_template
return self.parent.yaml_template
"""
if self.is_root:
if not self._model_yaml_template:
- with open(settings.MODEL_YAML_TEMPLATE, "r",
- encoding="utf-8") as model_yaml_template:
- self._model_yaml_template: dict = load(model_yaml_template, SafeLoader)
+ with open(
+ settings.MODEL_YAML_TEMPLATE, "r", encoding="utf-8"
+ ) as model_yaml_template:
+ self._model_yaml_template: dict = load(
+ model_yaml_template, SafeLoader
+ )
return self._model_yaml_template
return self.parent.model_yaml_template
"""Distribute service."""
super().execute()
with tracer.start_as_current_span(
- "sdc.service.distribute",
- attributes={"service.name": self.service_name}
+ "sdc.service.distribute", attributes={"service.name": self.service_name}
) as dist_span:
service: Service = Service.get_by_name(name=self.service_name)
if service:
if not service.distributed:
dist_span.set_attribute("service.already_distributed", False)
service.distribute()
- self._logger.info(f"Service {self.service_name} distributed successfully.")
+ self._logger.info(
+ f"Service {self.service_name} distributed successfully."
+ )
else:
dist_span.set_attribute("service.already_distributed", True)
- self._logger.info(f"Service {self.service_name} is already distributed.")
+ self._logger.info(
+ f"Service {self.service_name} is already distributed."
+ )
else:
dist_span.set_attribute("service.found", False)
- raise onap_test_exceptions.OnapTestException(f"Service {self.service_name} "
- f"not found for distribution.")
+ raise onap_test_exceptions.OnapTestException(
+ f"Service {self.service_name} " f"not found for distribution."
+ )
class BaseServiceDistributionComponentCheckStep(BaseStep):
service_model = None
- def __init__(self, component_name: str, break_on_error: bool = True, load_model: bool = True):
+ def __init__(
+ self, component_name: str, break_on_error: bool = True, load_model: bool = True
+ ):
"""Initialize step.
Args:
component_name (str): Name of tested component
break_on_error (bool): If step breaks execution when failed
"""
- super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP,
- break_on_error=break_on_error)
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP, break_on_error=break_on_error)
self.component_name = component_name
self.service: Service = None
self.load_model = load_model
"""
if cleanup:
return True
- return self.load_model or BaseServiceDistributionComponentCheckStep.service_model
+ return (
+ self.load_model or BaseServiceDistributionComponentCheckStep.service_model
+ )
def execute(self):
"""Check service distribution status."""
super().execute()
if not BaseServiceDistributionComponentCheckStep.service_model:
- BaseServiceDistributionComponentCheckStep.service_model = Service.get_by_name(
- name=settings.SERVICE_NAME)
+ BaseServiceDistributionComponentCheckStep.service_model = (
+ Service.get_by_name(name=settings.SERVICE_NAME)
+ )
self.service = BaseServiceDistributionComponentCheckStep.service_model
def _raise_reason(self, reason, exc=None):
"""Workaround to fix."""
@classmethod
- def get_all(cls,
- invariant_id: str = None,
- resource_version: str = None) -> Iterator["Model"]:
+ def get_all(
+ cls, invariant_id: str = None, resource_version: str = None
+ ) -> Iterator["Model"]:
"""Get all models.
Args:
"""
filter_parameters: dict = cls.filter_none_key_values(
- {"model-invariant-id": invariant_id,
- "resource-version": resource_version}
+ {
+ "model-invariant-id": invariant_id,
+ "resource-version": resource_version,
+ }
)
url: str = f"{cls.get_all_url()}?{urlencode(filter_parameters)}"
- for model in cls.send_message_json("GET", "Get A&AI sdc models",
- url).get("model", []):
+ for model in cls.send_message_json("GET", "Get A&AI sdc models", url).get(
+ "model", []
+ ):
yield Model(
invariant_id=model.get("model-invariant-id"),
model_type=model.get("model-type"),
- resource_version=model.get("resource-version")
+ resource_version=model.get("resource-version"),
)
def __init__(self):
"""Initialize step."""
BaseServiceDistributionComponentCheckStep.__init__(
- self, component_name="AAI", load_model=False)
+ self, component_name="AAI", load_model=False
+ )
@BaseStep.store_state
def execute(self):
super().execute()
try:
aai_services = self.ModelWithGet.get_all(
- invariant_id=self.service.invariant_uuid)
+ invariant_id=self.service.invariant_uuid
+ )
for aai_service in aai_services:
- self._logger.info(
- f"Resolved {aai_service.invariant_id} aai service")
+ self._logger.info(f"Resolved {aai_service.invariant_id} aai service")
except ResourceNotFound as e:
msg = "Service model is missing in AAI."
self._logger.error(msg)
def __init__(self):
"""Initialize step."""
BaseServiceDistributionComponentCheckStep.__init__(
- self, component_name="SDNC", load_model=False)
+ self, component_name="SDNC", load_model=False
+ )
@BaseStep.store_state
def execute(self):
super().execute()
if settings.IN_CLUSTER:
login, password = KubernetesHelper.get_credentials_from_secret(
- settings.SDNC_SECRET_NAME, self.SDNC_DB_LOGIN, self.SDNC_DB_PASSWORD)
+ settings.SDNC_SECRET_NAME, self.SDNC_DB_LOGIN, self.SDNC_DB_PASSWORD
+ )
conn = None
try:
conn = mysql.connect(
host=settings.SDNC_DB_PRIMARY_HOST,
port=settings.SDNC_DB_PORT,
user=login,
- password=password)
+ password=password,
+ )
cursor = conn.cursor()
cursor.execute(
- f"SELECT * FROM service_model WHERE service_uuid = '{self.service.uuid}';")
+ f"SELECT * FROM service_model WHERE service_uuid = '{self.service.uuid}';"
+ )
cursor.fetchall()
if cursor.rowcount <= 0:
msg = "Service model is missing in SDNC."
COMPONENTS_DISTRIBUTION_VERIFICATION_MAP = {
"AAI": VerifyServiceDistributionInAaiStep,
"SDNC": VerifyServiceDistributionInSdncStep,
- "SO": VerifyServiceDistributionInSoStep
+ "SO": VerifyServiceDistributionInSoStep,
}
def __init__(self):
"""Initialize step."""
super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self.add_step(ServiceDistributionWaitStep())
- modules = sorted(settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS,
- reverse=True)
+ modules = sorted(settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS, reverse=True)
for notified_module in modules:
component_name = notified_module.split("-")[0].upper()
- self.add_step(VerifyServiceDistributionStatusStep(
- notified_module=notified_module, component_name=component_name))
+ self.add_step(
+ VerifyServiceDistributionStatusStep(
+ notified_module=notified_module, component_name=component_name
+ )
+ )
try:
- self.add_step(self.COMPONENTS_DISTRIBUTION_VERIFICATION_MAP[component_name]())
+ self.add_step(
+ self.COMPONENTS_DISTRIBUTION_VERIFICATION_MAP[component_name]()
+ )
except KeyError:
pass
# under dedicated step anyway
if not step.is_executed:
raise onap_test_exceptions.ServiceDistributionException(
- "Service distribution has failed")
+ "Service distribution has failed"
+ )
class ServiceDistributionWaitStep(BaseServiceDistributionComponentCheckStep):
self._logger.info("******** Check Service Distribution *******")
distribution_completed = False
nb_try = 0
- while distribution_completed is False and \
- nb_try < settings.SERVICE_DISTRIBUTION_NUMBER_OF_TRIES:
+ while (
+ distribution_completed is False
+ and nb_try < settings.SERVICE_DISTRIBUTION_NUMBER_OF_TRIES
+ ):
distribution_completed = self.service.distributed
if distribution_completed is True:
self._logger.info(
"Service Distribution for %s is sucessfully finished",
- self.service.name)
+ self.service.name,
+ )
break
self._logger.info(
"Service Distribution for %s ongoing, Wait for %d s",
- self.service.name, settings.SERVICE_DISTRIBUTION_SLEEP_TIME)
+ self.service.name,
+ settings.SERVICE_DISTRIBUTION_SLEEP_TIME,
+ )
time.sleep(settings.SERVICE_DISTRIBUTION_SLEEP_TIME)
nb_try += 1
component_name (str): Name of the module's component
"""
super().__init__(
- component_name=component_name, break_on_error=False, load_model=False)
+ component_name=component_name, break_on_error=False, load_model=False
+ )
self.component_id = notified_module
@property
present = False
msg = ""
distributed = False
- for status in latest_distribution.distribution_status_list:
+ distribution_status_list = (
+ latest_distribution.distribution_status_list
+ if latest_distribution
+ else []
+ )
+ for status in distribution_status_list:
if status.component_id == self.component_id:
present = True
if status.distributed:
msg = f"Service model distribution to {self.component_id} \
was not completed"
else:
- msg = f"Service model was not distributed to {self.component_id}"
+ msg = (
+ f"Service model was not distributed to {self.component_id}"
+ )
self._raise_reason(msg)
msg = f"Service {self.service.name} is distributed in {self.component_name} \
and {self.component_id}."
--- /dev/null
+"""Test module for service onboarding functionality."""
+import sys
+from unittest import mock
+
+from onapsdk.configuration import settings
+from onapsdk.exceptions import ResourceNotFound
+from onapsdk.sdc2.sdc_resource import LifecycleState
+from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
+
+# Mock the kubernetes helper module before any imports to avoid settings dependency
+sys.modules['onaptests.utils.kubernetes'] = mock.MagicMock()
+
+# Create and configure settings mock before importing onapsdk
+settings.LOG_CONFIG = {
+ "version": 1,
+ "disable_existing_loggers": False,
+ "formatters": {
+ "default": {
+ "class": "logging.Formatter",
+ "format": "%(message)s"
+ }
+ },
+ "handlers": {
+ "console": {
+ "level": "DEBUG",
+ "class": "logging.StreamHandler",
+ "formatter": "default"
+ }
+ },
+ "root": {
+ "level": "DEBUG",
+ "handlers": ["console"]
+ }
+}
+settings.K8S_TESTS_NAMESPACE = 'test-namespace'
+settings.CLEANUP_FLAG = False
+settings.SERVICE_YAML_TEMPLATE = 'test.yaml'
+settings.MODEL_YAML_TEMPLATE = None
+settings.SDC_CLEANUP = False
+settings.IF_VALIDATION = False
+
+
+@mock.patch("onaptests.steps.onboard.service.YamlTemplateVfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.YamlTemplatePnfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.settings")
+@mock.patch("onaptests.steps.onboard.service.Service")
+@mock.patch("onaptests.steps.onboard.service.tracer")
+def test_service_distribute_called_for_new_service(
+ mock_tracer, mock_service_class, mock_settings, mock_pnf_step, mock_vf_step
+):
+ """Test that service.distribute() is called when creating a new service."""
+ # Setup mock settings
+ mock_settings.SERVICE_YAML_TEMPLATE = "test_service.yaml"
+ mock_settings.MODEL_YAML_TEMPLATE = None
+ mock_settings.CLEANUP_FLAG = False
+
+ # Setup mock service instance
+ mock_service_instance = mock.MagicMock()
+ mock_service_instance.name = "test_service"
+ mock_service_instance.lifecycle_state = LifecycleState.CERTIFIED
+ mock_service_instance.distributed = False
+
+ # Mock Service.get_by_name to raise ResourceNotFound (new service scenario)
+ mock_service_class.get_by_name.side_effect = ResourceNotFound
+
+ # Mock Service.create to return our mock service
+ mock_service_class.create.return_value = mock_service_instance
+
+ # Setup tracer context managers
+ mock_tracer.start_as_current_span.return_value.__enter__ = mock.MagicMock()
+ mock_tracer.start_as_current_span.return_value.__exit__ = mock.MagicMock()
+
+ # Mock the yaml template loading
+ mock_yaml_data = {
+ "test_service": {
+ "instantiation_type": "Macro"
+ }
+ }
+
+ with mock.patch("builtins.open", mock.mock_open(read_data="")):
+ with mock.patch("onaptests.steps.onboard.service.load", return_value=mock_yaml_data):
+ # Create and execute the step
+ step = YamlTemplateServiceOnboardStep()
+ step.execute()
+
+ # Verify that service.distribute() was called
+ mock_service_instance.distribute.assert_called_once()
+
+
+@mock.patch("onaptests.steps.onboard.service.YamlTemplateVfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.YamlTemplatePnfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.settings")
+@mock.patch("onaptests.steps.onboard.service.Service")
+@mock.patch("onaptests.steps.onboard.service.tracer")
+def test_service_distribute_called_for_not_certified_service(
+ mock_tracer, mock_service_class, mock_settings, mock_pnf_step, mock_vf_step
+):
+ """Test that service.distribute() is called after certifying a service."""
+ # Setup mock settings
+ mock_settings.SERVICE_YAML_TEMPLATE = "test_service.yaml"
+ mock_settings.MODEL_YAML_TEMPLATE = None
+ mock_settings.CLEANUP_FLAG = False
+
+ # Setup mock service instance in NOT_CERTIFIED_CHECKOUT state (returned by get_by_name)
+ mock_existing_service = mock.MagicMock()
+ mock_existing_service.name = "test_service"
+ mock_existing_service.lifecycle_state = LifecycleState.NOT_CERTIFIED_CHECKOUT
+ mock_existing_service.distributed = False
+
+ # Setup mock for newly created service (returned by Service.create)
+ mock_new_service = mock.MagicMock()
+ mock_new_service.name = "test_service"
+ mock_new_service.lifecycle_state = LifecycleState.NOT_CERTIFIED_CHECKOUT
+ mock_new_service.distributed = False
+
+ # Mock Service.get_by_name to return existing service
+ mock_service_class.get_by_name.return_value = mock_existing_service
+ # Mock Service.create to return new service instance
+ mock_service_class.create.return_value = mock_new_service
+
+ # Setup tracer context managers
+ mock_tracer.start_as_current_span.return_value.__enter__ = mock.MagicMock()
+ mock_tracer.start_as_current_span.return_value.__exit__ = mock.MagicMock()
+
+ # Mock the yaml template loading
+ mock_yaml_data = {
+ "test_service": {
+ "instantiation_type": "Macro"
+ }
+ }
+
+ with mock.patch("builtins.open", mock.mock_open(read_data="")):
+ with mock.patch("onaptests.steps.onboard.service.load", return_value=mock_yaml_data):
+ # Create and execute the step
+ step = YamlTemplateServiceOnboardStep()
+ step.execute()
+
+ # Verify that lifecycle_operation (CERTIFY) was called on new service
+ mock_new_service.lifecycle_operation.assert_called_once()
+
+ # Verify that service.distribute() was called on new service
+ mock_new_service.distribute.assert_called_once()
+
+
+@mock.patch("onaptests.steps.onboard.service.YamlTemplateVfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.YamlTemplatePnfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.settings")
+@mock.patch("onaptests.steps.onboard.service.Service")
+@mock.patch("onaptests.steps.onboard.service.tracer")
+def test_service_distribute_not_called_if_already_distributed(
+ mock_tracer, mock_service_class, mock_settings, mock_pnf_step, mock_vf_step
+):
+ """Test that execution returns early if service is already distributed."""
+ # Setup mock settings
+ mock_settings.SERVICE_YAML_TEMPLATE = "test_service.yaml"
+ mock_settings.MODEL_YAML_TEMPLATE = None
+ mock_settings.CLEANUP_FLAG = False
+
+ # Setup mock service instance that is already distributed
+ mock_service_instance = mock.MagicMock()
+ mock_service_instance.name = "test_service"
+ mock_service_instance.lifecycle_state = LifecycleState.CERTIFIED
+ mock_service_instance.distributed = True
+
+ # Mock Service.get_by_name to return existing distributed service
+ mock_service_class.get_by_name.return_value = mock_service_instance
+
+ # Setup tracer context managers
+ mock_tracer.start_as_current_span.return_value.__enter__ = mock.MagicMock()
+ mock_tracer.start_as_current_span.return_value.__exit__ = mock.MagicMock()
+
+ # Mock the yaml template loading
+ mock_yaml_data = {
+ "test_service": {
+ "instantiation_type": "Macro"
+ }
+ }
+
+ with mock.patch("builtins.open", mock.mock_open(read_data="")):
+ with mock.patch("onaptests.steps.onboard.service.load", return_value=mock_yaml_data):
+ # Create and execute the step
+ step = YamlTemplateServiceOnboardStep()
+ step.execute()
+
+ # Verify that service.distribute() was NOT called (already distributed)
+ mock_service_instance.distribute.assert_not_called()
+ # Verify that Service.create was NOT called
+ mock_service_class.create.assert_not_called()
+
+
+@mock.patch("onaptests.steps.onboard.service.YamlTemplateVfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.YamlTemplatePnfOnboardStep")
+@mock.patch("onaptests.steps.onboard.service.settings")
+@mock.patch("onaptests.steps.onboard.service.Service")
+@mock.patch("onaptests.steps.onboard.service.tracer")
+def test_service_distribute_called_for_certified_not_distributed_service(
+ mock_tracer, mock_service_class, mock_settings, mock_pnf_step, mock_vf_step
+):
+ """Test that service.distribute() is called for a certified service."""
+ # Setup mock settings
+ mock_settings.SERVICE_YAML_TEMPLATE = "test_service.yaml"
+ mock_settings.MODEL_YAML_TEMPLATE = None
+ mock_settings.CLEANUP_FLAG = False
+
+ # Setup mock for existing service (returned by get_by_name)
+ mock_existing_service = mock.MagicMock()
+ mock_existing_service.name = "test_service"
+ mock_existing_service.lifecycle_state = LifecycleState.CERTIFIED
+ mock_existing_service.distributed = False
+
+ # Setup mock for newly created service (returned by Service.create)
+ mock_new_service = mock.MagicMock()
+ mock_new_service.name = "test_service"
+ mock_new_service.lifecycle_state = LifecycleState.CERTIFIED
+ mock_new_service.distributed = False
+
+ # Mock Service.get_by_name to return existing certified service
+ mock_service_class.get_by_name.return_value = mock_existing_service
+ # Mock Service.create to return new service instance
+ mock_service_class.create.return_value = mock_new_service
+
+ # Setup tracer context managers
+ mock_tracer.start_as_current_span.return_value.__enter__ = mock.MagicMock()
+ mock_tracer.start_as_current_span.return_value.__exit__ = mock.MagicMock()
+
+ # Mock the yaml template loading
+ mock_yaml_data = {
+ "test_service": {
+ "instantiation_type": "Macro"
+ }
+ }
+
+ with mock.patch("builtins.open", mock.mock_open(read_data="")):
+ with mock.patch("onaptests.steps.onboard.service.load", return_value=mock_yaml_data):
+ # Create and execute the step
+ step = YamlTemplateServiceOnboardStep()
+ step.execute()
+
+ # Verify that lifecycle_operation was NOT called (already certified)
+ mock_new_service.lifecycle_operation.assert_not_called()
+
+ # Verify that service.distribute() WAS called on new service
+ mock_new_service.distribute.assert_called_once()
Code Review / testsuite / pythonsdk-tests.git / commitdiff