feat(pluggables): ECOC2026 - Enhance KPI descriptor and collector request... (57fcefb7) · Commits · TFS / controller

src/tests/ecoc26_pluggables/descriptors/ecoc26_messages.py

+94 −32

Original line number	Diff line number	Diff line
		@@ -12,7 +12,8 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import uuid, random
		import os, uuid, random
		from uuid import UUID
		from common.DeviceTypes import DeviceTypeEnum
		from common.proto import kpi_manager_pb2
		from common.proto import telemetry_frontend_pb2
		@@ -20,36 +21,73 @@ from common.proto.kpi_sample_types_pb2 import KpiSampleType
		from common.proto.context_pb2 import DeviceDriverEnum

		# create UUIDs (for both Received Power and PRE-FEC BER pluggable input and output KPIs)
		KPI_DESCRIPTOR_RECEIVED_POWER_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_RECEIVED_POWER_HUB_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_RECEIVED_POWER_LEAF_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_RECEIVED_POWER_UUID = KPI_DESCRIPTOR_RECEIVED_POWER_LEAF_UUID
		KPI_DESCRIPTOR_PRE_FEC_BER_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_AGG_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_HUB_AGG_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_LEAF_AGG_UUID = str(uuid.uuid4())
		KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_AGG_UUID = KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_LEAF_AGG_UUID
		KPI_DESCRIPTOR_PRE_FEC_BER_AGG_UUID = str(uuid.uuid4())

		# create collector UUIDs later ...
		COLLECTOR_RECEIVED_POWER_UUID = str(uuid.uuid4())
		COLLECTOR_RECEIVED_POWER_HUB_UUID = str(uuid.uuid4())
		COLLECTOR_RECEIVED_POWER_LEAF_UUID = str(uuid.uuid4())
		COLLECTOR_RECEIVED_POWER_UUID = COLLECTOR_RECEIVED_POWER_LEAF_UUID
		COLLECTOR_PRE_FEC_BER_UUID = str(uuid.uuid4())

		# Router Device UUID (used in KPI Descriptor creation)
		ROUTER_HUB_DEVICE_UUID = "a79e0852-a603-5200-8de4-aeef1059c27f" # IP1
		ROUTER_LEAF_DEVICE_UUID = "ad768743-bafd-598e-b256-de435bee50c2" # IP2
		# Device IDs and optical channels used in KPI descriptor creation.
		# The real ECOC26 descriptor uses r1/r2; the old mock descriptor used IP1/IP2.
		_TFS_NAMESPACE_UUID = UUID('200e3a1f-2223-534f-a100-758e29c37f40')
		ROUTER_HUB_DEVICE_ID = os.environ.get('ECOC26_HUB_DEVICE_ID', 'r1')
		ROUTER_LEAF_DEVICE_ID = os.environ.get('ECOC26_LEAF_DEVICE_ID', 'r2')
		ROUTER_HUB_CHANNEL = os.environ.get('ECOC26_HUB_OPTICAL_CHANNEL', 'channel-5')
		ROUTER_LEAF_CHANNEL = os.environ.get('ECOC26_LEAF_OPTICAL_CHANNEL', 'channel-5')

		def create_kpi_descriptor_PRE_FEC_BER_request():
		def _tfs_uuid(str_uuid_or_name: str) -> str:
		try:
		return str(UUID(str_uuid_or_name))
		except (TypeError, ValueError, AttributeError):
		return str(uuid.uuid5(_TFS_NAMESPACE_UUID, str_uuid_or_name))

		ROUTER_HUB_DEVICE_UUID = _tfs_uuid(ROUTER_HUB_DEVICE_ID)
		ROUTER_LEAF_DEVICE_UUID = _tfs_uuid(ROUTER_LEAF_DEVICE_ID)

		def _create_kpi_descriptor_request(kpi_uuid, description, sample_type, device_uuid, endpoint_uuid):
		_create_kpi_request = kpi_manager_pb2.KpiDescriptor()
		_create_kpi_request.kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_PRE_FEC_BER_UUID
		_create_kpi_request.kpi_description = "Pluggable KPI PRE-FEC BER descriptor for testing purposes"
		_create_kpi_request.kpi_sample_type = KpiSampleType.KPISAMPLETYPE_PRE_FEC_BER_PLUGGABLE
		_create_kpi_request.device_id.device_uuid.uuid = ROUTER_LEAF_DEVICE_UUID
		_create_kpi_request.endpoint_id.endpoint_uuid.uuid = "channel-1"
		_create_kpi_request.kpi_id.kpi_id.uuid = kpi_uuid
		_create_kpi_request.kpi_description = description
		_create_kpi_request.kpi_sample_type = sample_type
		_create_kpi_request.device_id.device_uuid.uuid = device_uuid
		_create_kpi_request.endpoint_id.endpoint_uuid.uuid = endpoint_uuid
		return _create_kpi_request

		def create_kpi_descriptor_PRE_FEC_BER_request():
		return _create_kpi_descriptor_request(
		KPI_DESCRIPTOR_PRE_FEC_BER_UUID,
		"Pluggable KPI PRE-FEC BER descriptor for testing purposes",
		KpiSampleType.KPISAMPLETYPE_PRE_FEC_BER_PLUGGABLE,
		ROUTER_LEAF_DEVICE_UUID,
		ROUTER_LEAF_CHANNEL)

		def create_kpi_descriptor_RECEIVED_POWER_hub_request():
		return _create_kpi_descriptor_request(
		KPI_DESCRIPTOR_RECEIVED_POWER_HUB_UUID,
		"Hub pluggable KPI Received power descriptor for testing purposes",
		KpiSampleType.KPISAMPLETYPE_RECEIVED_POWER_PLUGGABLE,
		ROUTER_HUB_DEVICE_UUID,
		ROUTER_HUB_CHANNEL)

		def create_kpi_descriptor_RECEIVED_POWER_leaf_request():
		return _create_kpi_descriptor_request(
		KPI_DESCRIPTOR_RECEIVED_POWER_LEAF_UUID,
		"Leaf pluggable KPI Received power descriptor for testing purposes",
		KpiSampleType.KPISAMPLETYPE_RECEIVED_POWER_PLUGGABLE,
		ROUTER_LEAF_DEVICE_UUID,
		ROUTER_LEAF_CHANNEL)

		def create_kpi_descriptor_RECEIVED_POWER_request():
		_create_kpi_request = kpi_manager_pb2.KpiDescriptor()
		_create_kpi_request.kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_RECEIVED_POWER_UUID
		_create_kpi_request.kpi_description = "Pluggable KPI Received power descriptor for testing purposes"
		_create_kpi_request.kpi_sample_type = KpiSampleType.KPISAMPLETYPE_RECEIVED_POWER_PLUGGABLE
		_create_kpi_request.device_id.device_uuid.uuid = ROUTER_LEAF_DEVICE_UUID
		_create_kpi_request.endpoint_id.endpoint_uuid.uuid = "channel-1"
		return _create_kpi_request
		return create_kpi_descriptor_RECEIVED_POWER_leaf_request()

		def create_PRE_FEC_BER_kpi_id_request():
		_create_kpi_id = kpi_manager_pb2.KpiId()
		@@ -61,6 +99,16 @@ def create_RECEIVED_POWER_kpi_id_request():
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_RECEIVED_POWER_UUID
		return _create_kpi_id

		def create_RECEIVED_POWER_hub_kpi_id_request():
		_create_kpi_id = kpi_manager_pb2.KpiId()
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_RECEIVED_POWER_HUB_UUID
		return _create_kpi_id

		def create_RECEIVED_POWER_leaf_kpi_id_request():
		_create_kpi_id = kpi_manager_pb2.KpiId()
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_RECEIVED_POWER_LEAF_UUID
		return _create_kpi_id


		# ── Aggregated Output KPI IDs (no descriptor needed — just UUIDs) ────────────

		@@ -69,6 +117,16 @@ def create_RECEIVED_POWER_AGG_OUTPUT_kpi_id_request():
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_AGG_UUID
		return _create_kpi_id

		def create_RECEIVED_POWER_AGG_OUTPUT_hub_kpi_id_request():
		_create_kpi_id = kpi_manager_pb2.KpiId()
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_HUB_AGG_UUID
		return _create_kpi_id

		def create_RECEIVED_POWER_AGG_OUTPUT_leaf_kpi_id_request():
		_create_kpi_id = kpi_manager_pb2.KpiId()
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_LEAF_AGG_UUID
		return _create_kpi_id

		def create_PRE_FEC_BER_AGG_OUTPUT_kpi_id_request():
		_create_kpi_id = kpi_manager_pb2.KpiId()
		_create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_PRE_FEC_BER_AGG_UUID
		@@ -76,22 +134,26 @@ def create_PRE_FEC_BER_AGG_OUTPUT_kpi_id_request():

		# TELEMETRY_COLLECTOR REQUESTS MESAGES

		def create_collector_request_received_power():
		def _create_collector_request(collector_uuid, kpi_uuid):
		_create_collector_request = telemetry_frontend_pb2.Collector()
		_create_collector_request.collector_id.collector_id.uuid = COLLECTOR_RECEIVED_POWER_UUID
		_create_collector_request.kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_RECEIVED_POWER_UUID
		_create_collector_request.collector_id.collector_id.uuid = collector_uuid
		_create_collector_request.kpi_id.kpi_id.uuid = kpi_uuid
		_create_collector_request.duration_s = float(random.randint(80, 90))
		_create_collector_request.interval_s = float(random.randint(6,8))
		_create_collector_request.coll_meta_info.device_driver = DeviceDriverEnum.DEVICEDRIVER_NETCONF_OC_PLUGGABLE
		_create_collector_request.coll_meta_info.device_type = DeviceTypeEnum.PACKET_ROUTER.value
		return _create_collector_request

		def create_collector_request_received_power_hub():
		return _create_collector_request(
		COLLECTOR_RECEIVED_POWER_HUB_UUID, KPI_DESCRIPTOR_RECEIVED_POWER_HUB_UUID)

		def create_collector_request_received_power_leaf():
		return _create_collector_request(
		COLLECTOR_RECEIVED_POWER_LEAF_UUID, KPI_DESCRIPTOR_RECEIVED_POWER_LEAF_UUID)

		def create_collector_request_received_power():
		return create_collector_request_received_power_leaf()

		def create_collector_request_PRE_FEC_BER():
		_create_collector_request = telemetry_frontend_pb2.Collector()
		_create_collector_request.collector_id.collector_id.uuid = COLLECTOR_PRE_FEC_BER_UUID
		_create_collector_request.kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_PRE_FEC_BER_UUID
		_create_collector_request.duration_s = float(random.randint(80, 90))
		_create_collector_request.interval_s = float(random.randint(6,8))
		_create_collector_request.coll_meta_info.device_driver = DeviceDriverEnum.DEVICEDRIVER_NETCONF_OC_PLUGGABLE
		_create_collector_request.coll_meta_info.device_type = DeviceTypeEnum.PACKET_ROUTER.value
		return _create_collector_request
		return _create_collector_request(COLLECTOR_PRE_FEC_BER_UUID, KPI_DESCRIPTOR_PRE_FEC_BER_UUID)

src/tests/ecoc26_pluggables/helper_methods/add_topology.py

+8 −1

Original line number	Diff line number	Diff line
		@@ -28,6 +28,13 @@ from src.tests.ecoc26_pluggables.descriptors.ecoc26_messages import (


		LOGGER = logging.getLogger(__name__)
		_DEVICE_IDS_TO_REMOVE = {
		ROUTER_HUB_DEVICE_UUID,
		ROUTER_LEAF_DEVICE_UUID,
		# Real descriptor names. Context accepts names and resolves them to UUIDs.
		"r1",
		"r2",
		}
		# =======================
		# Setup Topology and Devices
		# =======================
		@@ -64,7 +71,7 @@ def load_topology(
		# ----- Remove Topology -----
		def _remove_devices_topology_context(context_client):
		# Remove devices first — their endpoints reference the topology via FK
		for device_uuid in [ROUTER_HUB_DEVICE_UUID, ROUTER_LEAF_DEVICE_UUID]:
		for device_uuid in _DEVICE_IDS_TO_REMOVE:
		try:
		dev_id = DeviceId()
		dev_id.device_uuid.uuid = device_uuid

src/tests/ecoc26_pluggables/run_in_K8s/run_zsm_test.sh

+3 −2

Original line number	Diff line number	Diff line
		@@ -19,8 +19,9 @@
		# Tests executed:
		# test_zsm_create_with_failure_notification
		# – loads topology, creates KPI descriptor, creates L3NM + OPTICAL_CONNECTIVITY
		# services, builds ZSMCreateRequest with GENERATE_FAILURE_NOTIFICATION action,
		# invokes automation_client.ZSMCreate(), asserts response.
		# services, starts hub/leaf received-power collectors, verifies KPI samples,
		# builds ZSMCreateRequest with GENERATE_FAILURE_NOTIFICATION action, invokes
		# automation_client.ZSMCreate(), asserts response.
		#
		# Usage:
		# ./run_zsm_test.sh

src/tests/ecoc26_pluggables/run_in_K8s/zsm_test_job.yaml

+18 −2

Original line number	Diff line number	Diff line
		@@ -64,9 +64,25 @@ spec:
		# 3. /home/cttc/tfs-ctrl → src.tests.ecoc26_pluggables.*
		- name: PYTHONPATH
		value: "/var/teraflow:/home/cttc/tfs-ctrl/src:/home/cttc/tfs-ctrl"
		# Use the automation-specific topology descriptor
		# Use the real data-plane topology descriptor
		- name: TFS_TOPOLOGY_DESCRIPTOR
		value: "ecoc26_topology_automation.json"
		value: "ecoc26_topo_devices_actual_data_plane.json"
		# Real ECOC26 data-plane mapping. Device IDs are resolved by
		# Context into deterministic UUIDs before KPI descriptors use them.
		- name: ECOC26_HUB_DEVICE_ID
		value: "r1"
		- name: ECOC26_LEAF_DEVICE_ID
		value: "r2"
		- name: ECOC26_HUB_OPTICAL_CHANNEL
		value: "channel-5"
		- name: ECOC26_LEAF_OPTICAL_CHANNEL
		value: "channel-5"
		# Expected real received-power ranges. These checks ensure the hub
		# and leaf collectors are both producing samples, not just KPI IDs.
		- name: ECOC26_HUB_RECEIVED_POWER_RANGE
		value: "-80,-30"
		- name: ECOC26_LEAF_RECEIVED_POWER_RANGE
		value: "-30,0"
		volumeMounts:
		- name: tfs-source
		mountPath: /home/cttc/tfs-ctrl

src/tests/ecoc26_pluggables/tests/test_zsm_failure_notification.py

+149 −37

Original line number	Diff line number	Diff line
		@@ -12,7 +12,7 @@
		# See the License for the specific language governing permissions and
		# limitations under the License.

		import json, logging, uuid
		import json, logging, os, time, uuid

		from common.proto.analytics_frontend_pb2 import AnalyzerOperationMode
		from common.proto.automation_pb2 import ZSMCreateRequest, ZSMService
		@@ -27,10 +27,14 @@ from common.tools.context_queries.Device import get_device
		from common.tools.context_queries.Topology import get_topology
		from src.tests.ecoc26_pluggables.descriptors.ecoc26_messages import (
		ROUTER_HUB_DEVICE_UUID, ROUTER_LEAF_DEVICE_UUID,
		create_RECEIVED_POWER_kpi_id_request,
		create_RECEIVED_POWER_AGG_OUTPUT_kpi_id_request,
		create_kpi_descriptor_RECEIVED_POWER_request,
		create_collector_request_received_power)
		create_RECEIVED_POWER_hub_kpi_id_request,
		create_RECEIVED_POWER_leaf_kpi_id_request,
		create_RECEIVED_POWER_AGG_OUTPUT_hub_kpi_id_request,
		create_RECEIVED_POWER_AGG_OUTPUT_leaf_kpi_id_request,
		create_kpi_descriptor_RECEIVED_POWER_hub_request,
		create_kpi_descriptor_RECEIVED_POWER_leaf_request,
		create_collector_request_received_power_hub,
		create_collector_request_received_power_leaf)
		from src.tests.ecoc26_pluggables.helper_methods.Fixtuers import (
		automation_client, context_client, device_client, kpi_manager_client,
		telemetry_frontend_client)
		@@ -40,12 +44,49 @@ from src.tests.ecoc26_pluggables.helper_methods.add_topology import (
		LOGGER = logging.getLogger(__name__)
		TARGET_SERVICE_UUID = str(uuid.uuid4())
		TELEMETRY_SERVICE_UUID = str(uuid.uuid4())
		_SAMPLE_TIMEOUT_S = float(os.environ.get("ECOC26_ZSM_SAMPLE_TIMEOUT_S", "45"))

		# Hardcoded UUIDs from the topology descriptor (ecoc26_topology_automation.json)
		def _parse_power_range(env_name, default_range):
		raw_range = os.environ.get(env_name)
		if raw_range is None:
		return default_range
		try:
		low, high = [float(value.strip()) for value in raw_range.split(",", 1)]
		return low, high
		except ValueError as exc:
		raise ValueError(
		"{:s} must use '<min>,<max>' format, got {!r}".format(env_name, raw_range)
		) from exc

		_EXPECTED_RECEIVED_POWER_KPI_RANGES = {
		create_RECEIVED_POWER_hub_kpi_id_request().kpi_id.uuid: (
		"hub",
		_parse_power_range("ECOC26_HUB_RECEIVED_POWER_RANGE", (-80.0, -30.0)),
		),
		create_RECEIVED_POWER_leaf_kpi_id_request().kpi_id.uuid: (
		"leaf",
		_parse_power_range("ECOC26_LEAF_RECEIVED_POWER_RANGE", (-30.0, 0.0)),
		),
		}

		# Deterministic UUIDs produced by Context from descriptor names admin/admin.
		_CONTEXT_UUID = "43813baf-195e-5da6-af20-b3d0922e71a7"
		_TOPOLOGY_UUID = "c76135e3-24a8-5e92-9bed-c3c9139359c8"
		_EXPECTED_DEVICE_UUIDS = {ROUTER_HUB_DEVICE_UUID, ROUTER_LEAF_DEVICE_UUID}

		_RECEIVED_POWER_KPIS = (
		(
		"hub",
		create_RECEIVED_POWER_hub_kpi_id_request,
		create_kpi_descriptor_RECEIVED_POWER_hub_request,
		),
		(
		"leaf",
		create_RECEIVED_POWER_leaf_kpi_id_request,
		create_kpi_descriptor_RECEIVED_POWER_leaf_request,
		),
		)


		# ═══════════════════════════════════════════════════════════════════════════════
		# Test 1: Topology — ensure expected context, topology and devices exist
		@@ -86,18 +127,17 @@ def test_01_ensure_topology(context_client, device_client):
		# ═══════════════════════════════════════════════════════════════════════════════

		def test_02_ensure_kpi_descriptor(kpi_manager_client):
		"""Check if the RECEIVED_POWER KPI descriptor exists. Create it if missing."""
		"""Check if HUB/LEAF RECEIVED_POWER KPI descriptors exist. Create missing ones."""
		LOGGER.info(' >>> test_02_ensure_kpi_descriptor START: <<< ')
		for role, kpi_id_factory, descriptor_factory in _RECEIVED_POWER_KPIS:
		try:
		kpi_desc = kpi_manager_client.GetKpiDescriptor(
		create_RECEIVED_POWER_kpi_id_request())
		LOGGER.info('KPI Descriptor already exists: %s', str(kpi_desc))
		kpi_desc = kpi_manager_client.GetKpiDescriptor(kpi_id_factory())
		LOGGER.info('%s KPI Descriptor already exists: %s', role, str(kpi_desc))
		assert isinstance(kpi_desc, KpiDescriptor)
		except Exception:
		LOGGER.info('KPI Descriptor not found, creating...')
		response = kpi_manager_client.SetKpiDescriptor(
		create_kpi_descriptor_RECEIVED_POWER_request())
		LOGGER.info('SetKpiDescriptor response: %s', str(response))
		LOGGER.info('%s KPI Descriptor not found, creating...', role)
		response = kpi_manager_client.SetKpiDescriptor(descriptor_factory())
		LOGGER.info('%s SetKpiDescriptor response: %s', role, str(response))
		assert isinstance(response, KpiId)


		@@ -106,15 +146,22 @@ def test_02_ensure_kpi_descriptor(kpi_manager_client):
		# ═══════════════════════════════════════════════════════════════════════════════

		def test_03_start_collector(telemetry_frontend_client):
		"""Start the telemetry collector to stream RECEIVED_POWER metrics from the
		NETCONF pluggable device. The analyzer (started by the ZSM plugin in the
		next test) subscribes to this KPI stream to evaluate the threshold."""
		"""Start telemetry collectors to stream HUB/LEAF RECEIVED_POWER metrics.
		The analyzer (started by the ZSM plugin in the next test) subscribes to
		these KPI streams to evaluate the threshold."""
		LOGGER.info(' >>> test_03_start_collector START: <<< ')
		response = telemetry_frontend_client.StartCollector(
		create_collector_request_received_power())
		for role, request_factory in (
		("hub", create_collector_request_received_power_hub),
		("leaf", create_collector_request_received_power_leaf),
		):
		response = telemetry_frontend_client.StartCollector(request_factory())
		LOGGER.debug(str(response))
		assert isinstance(response, CollectorId)
		LOGGER.info('Collector started: %s', str(response.collector_id.uuid))
		LOGGER.info('%s collector started: %s', role, str(response.collector_id.uuid))
		_wait_for_kpi_samples(
		_EXPECTED_RECEIVED_POWER_KPI_RANGES,
		timeout_s=_SAMPLE_TIMEOUT_S,
		)


		# ═══════════════════════════════════════════════════════════════════════════════
		@@ -136,6 +183,7 @@ def test_04_zsm_create_with_failure_notification(
		target_service = _create_service(ServiceTypeEnum.SERVICETYPE_L3NM, TARGET_SERVICE_UUID)
		telemetry_service = _create_service(ServiceTypeEnum.SERVICETYPE_OPTICAL_CONNECTIVITY, TELEMETRY_SERVICE_UUID)
		created_services = []
		policy_rule_id = None

		try:
		for service in [target_service, telemetry_service]:
		@@ -157,6 +205,7 @@ def test_04_zsm_create_with_failure_notification(
		finally:
		# DO NOT add in actual integration: It is only for cleanup in tests.
		try:
		if policy_rule_id is not None:
		policy_client = PolicyClient()
		policy_client.PolicyDelete(policy_rule_id)
		policy_client.close()
		@@ -181,6 +230,61 @@ def _create_service(service_type, service_uuid=None):
		return service


		def _wait_for_kpi_samples(expected_kpis, timeout_s):
		from confluent_kafka import Consumer as KafkaConsumer, KafkaError
		from common.tools.kafka.Variables import KafkaConfig, KafkaTopic

		expected_kpi_ids = set(expected_kpis)
		consumer = KafkaConsumer({
		'bootstrap.servers': KafkaConfig.get_kafka_address(),
		'group.id': 'ecoc26-zsm-test-{:s}'.format(str(uuid.uuid4())),
		# Use unique KPI IDs per test process, so reading from the beginning is
		# safer than missing samples produced before partition assignment.
		'auto.offset.reset': 'earliest',
		})
		consumer.subscribe([KafkaTopic.VALUE.value])

		deadline = time.time() + timeout_s
		samples = {}
		out_of_range_samples = {}
		try:
		while time.time() < deadline and expected_kpi_ids.difference(samples):
		msg = consumer.poll(1.0)
		if msg is None:
		continue
		if msg.error():
		if msg.error().code() == KafkaError._PARTITION_EOF:
		continue
		raise RuntimeError('Kafka error while waiting for KPI samples: {:s}'.format(str(msg.error())))

		payload = json.loads(msg.value().decode('utf-8'))
		kpi_id = payload.get('kpi_id')
		if kpi_id in expected_kpi_ids:
		role, (min_value, max_value) = expected_kpis[kpi_id]
		try:
		kpi_value = float(payload.get('kpi_value'))
		except (TypeError, ValueError):
		out_of_range_samples[kpi_id] = payload
		LOGGER.warning('Received non-numeric %s KPI sample for %s: %s', role, kpi_id, payload)
		continue
		if not min_value <= kpi_value <= max_value:
		out_of_range_samples[kpi_id] = payload
		LOGGER.warning(
		'Received out-of-range %s KPI sample for %s: %s not in [%s, %s]',
		role, kpi_id, kpi_value, min_value, max_value)
		continue
		samples[kpi_id] = payload
		LOGGER.info('Received in-range %s KPI sample for %s: %s', role, kpi_id, payload)
		finally:
		consumer.close()

		missing_kpi_ids = expected_kpi_ids.difference(samples)
		assert not missing_kpi_ids, (
		'Timed out after {:.1f}s waiting for in-range received-power KPI samples: {}; '
		'last out-of-range samples: {}'.format(
		timeout_s, sorted(missing_kpi_ids), out_of_range_samples))


		def _create_zsm_create_request(target_service, telemetry_service):
		request = ZSMCreateRequest()
		request.target_service_id.CopyFrom(target_service.service_id)
		@@ -188,18 +292,25 @@ def _create_zsm_create_request(target_service, telemetry_service):

		request.analyzer.analyzer_id.analyzer_id.uuid = str(uuid.uuid4())
		request.analyzer.operation_mode = AnalyzerOperationMode.ANALYZEROPERATIONMODE_STREAMING
		input_kpi_ids = [
		create_RECEIVED_POWER_hub_kpi_id_request(),
		create_RECEIVED_POWER_leaf_kpi_id_request(),
		]
		output_kpi_ids = [
		create_RECEIVED_POWER_AGG_OUTPUT_hub_kpi_id_request(),
		create_RECEIVED_POWER_AGG_OUTPUT_leaf_kpi_id_request(),
		]

		request.analyzer.parameters["thresholds"] = json.dumps({
		"task_type": "AggregationHandler",
		"task_parameter": [{"last": [-30, -5]}],
		"task_parameter": [{"last": [-30, -5]} for _ in input_kpi_ids],
		})
		request.analyzer.parameters["window_size"] = "10"
		request.analyzer.parameters["window_slider"] = "5"
		request.analyzer.batch_min_duration_s = 10
		request.analyzer.batch_min_size = 5
		request.analyzer.input_kpi_ids.append(create_RECEIVED_POWER_kpi_id_request())

		output_kpi_id = create_RECEIVED_POWER_AGG_OUTPUT_kpi_id_request()
		request.analyzer.output_kpi_ids.append(output_kpi_id)
		request.analyzer.input_kpi_ids.extend(input_kpi_ids)
		request.analyzer.output_kpi_ids.extend(output_kpi_ids)

		request.policy.serviceId.CopyFrom(target_service.service_id)
		request.policy.policyRuleBasic.policyRuleId.uuid.uuid = str(uuid.uuid4())
		@@ -211,6 +322,7 @@ def _create_zsm_create_request(target_service, telemetry_service):
		action.action = PolicyRuleActionEnum.POLICY_RULE_ACTION_GENERATE_FAILURE_NOTIFICATION
		request.policy.policyRuleBasic.actionList.append(action)

		for output_kpi_id in output_kpi_ids:
		policy_kpi_id = PolicyRuleKpiId()
		policy_kpi_id.policyRuleKpiUuid.uuid = output_kpi_id.kpi_id.uuid
		request.policy.policyRuleBasic.policyRuleKpiList.append(policy_kpi_id)