Commit 40b0fc60 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

feat: ECOC 2026 - Pluggable Demo Test Files 

- Integrate ZSM failure notification tests
- Added creation methods of KPI, Telemtery, Automation
parent 00e6cf22

src/tests/ecoc26_pluggables/descriptors/ecoc26_messages.py

+21 −6
Original line number Diff line number Diff line
@@ -19,8 +19,11 @@ from common.proto import telemetry_frontend_pb2 
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_PRE_FEC_BER_UUID               = str(uuid.uuid4())

KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_AGG_UUID = str(uuid.uuid4())

KPI_DESCRIPTOR_PRE_FEC_BER_AGG_UUID           = str(uuid.uuid4())



# create collector UUIDs later ...

COLLECTOR_RECEIVED_POWER_UUID = str(uuid.uuid4())
@@ -59,14 +62,26 @@ def create_RECEIVED_POWER_kpi_id_request(): 
    return _create_kpi_id





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



def create_RECEIVED_POWER_AGG_OUTPUT_kpi_id_request():

    _create_kpi_id = kpi_manager_pb2.KpiId()

    _create_kpi_id.kpi_id.uuid = KPI_DESCRIPTOR_OUTPUT_RECEIVED_POWER_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

    return _create_kpi_id



# TELEMETRY_COLLECTOR REQUESTS MESAGES



def create_collector_request_received_power():

    _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.duration_s                     = float(random.randint(30, 50))

    _create_collector_request.interval_s                     = float(random.randint(2, 4))

    _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
@@ -75,8 +90,8 @@ 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(30, 50))

    _create_collector_request.interval_s                     = float(random.randint(2, 4))

    _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

src/tests/ecoc26_pluggables/helper_methods/Fixtuers.py

+9 −0
Original line number Diff line number Diff line
@@ -16,6 +16,7 @@ 
import pytest

import logging



from automation.client.AutomationClient      import AutomationClient

from context.client.ContextClient        import ContextClient

from device.client.DeviceClient          import DeviceClient

from kpi_manager.client.KpiManagerClient import KpiManagerClient
@@ -69,3 +70,11 @@ def telemetry_frontend_client(): 
    yield _client

    LOGGER.info('Closed TelemetryFrontendClient...')

    _client.close()



@pytest.fixture(scope='session')

def automation_client():

    _client = AutomationClient()

    LOGGER.info('Yielding Connected AutomationClient...')

    yield _client

    LOGGER.info('Closed AutomationClient...')

    _client.close()

src/tests/ecoc26_pluggables/helper_methods/add_topology.py

+4 −1
Original line number Diff line number Diff line
@@ -56,7 +56,7 @@ def load_topology( 




# ----- Remove Topology -----

def test_remove_topology(context_client):

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]:

        try:
@@ -76,6 +76,9 @@ def test_remove_topology(context_client): 
    response = context_client.RemoveTopology(topology_id)

    LOGGER.info(f"Topology removed: {response}")



    response = context_client.RemoveContext(context_id)

    LOGGER.info(f"Context removed: {response}")



# ----- Remove context -----

def test_remove_context(context_client):

    context_id = ContextId()

src/tests/ecoc26_pluggables/run_in_K8s/run_zsm_test.sh

0 → 100755
+81 −0
Original line number Diff line number Diff line 
#!/bin/bash

# Copyright 2022-2026 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

#

# Deploys and streams output from the ZSM failure notification pytest Job

# inside the tfs namespace.

#

# 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.

#

# Usage:

#   ./run_zsm_test.sh



set -euo pipefail



NAMESPACE="tfs"

JOB_NAME="zsm-automation-test"

MANIFEST="$(dirname "$0")/zsm_test_job.yaml"

LOG_FILE="$(dirname "$0")/../logs/zsm_test_$(date '+%Y%m%d_%H%M').log"



# ── Clean up any previous run ────────────────────────────────────────────────

if kubectl get job "${JOB_NAME}" -n "${NAMESPACE}" &>/dev/null; then

    kubectl delete job "${JOB_NAME}" -n "${NAMESPACE}" --wait=true >>"${LOG_FILE}" 2>&1

fi



# ── Submit the Job ───────────────────────────────────────────────────────────

kubectl apply -f "${MANIFEST}" >>"${LOG_FILE}" 2>&1



# ── Wait for the container to be Running (past ContainerCreating) ─────────────

until kubectl get pods -n "${NAMESPACE}" -l job-name="${JOB_NAME}" \

        --no-headers 2>/dev/null | grep -qE 'Running|Completed|Succeeded|Error'; do

    sleep 1

done



# ── Stream logs to file in the background as they are generated ──────────────

kubectl logs --follow -n "${NAMESPACE}" \

    -l job-name="${JOB_NAME}" \

    --tail=-1 \

    --pod-running-timeout=60s >>"${LOG_FILE}" 2>&1 &

LOGS_PID=$!



# ── Poll for job completion (Complete or Failed) ──────────────────────────────

for i in $(seq 1 60); do

    JOB_COND=$(kubectl get job "${JOB_NAME}" -n "${NAMESPACE}" \

        -o jsonpath='{.status.conditions[0].type}' 2>/dev/null || echo "")

    if [ "${JOB_COND}" = "Complete" ] || [ "${JOB_COND}" = "Failed" ]; then

        break

    fi

    sleep 3

done



# ── Wait for the log streamer to flush and exit ───────────────────────────────

wait "${LOGS_PID}" 2>/dev/null || true



# ── Capture status, then immediately delete the Job and its Pod ───────────────

JOB_STATUS=$(kubectl get job "${JOB_NAME}" -n "${NAMESPACE}" \

    -o jsonpath='{.status.conditions[0].type}' 2>/dev/null || echo "Unknown")

kubectl delete job "${JOB_NAME}" -n "${NAMESPACE}" --wait=false >>"${LOG_FILE}" 2>&1 || true



# ── Report final status ──────────────────────────────────────────────────────

if [ "${JOB_STATUS}" = "Complete" ]; then

    echo "[✓] Tests PASSED  — full log: ${LOG_FILE}"

    exit 0

else

    echo "[✗] Tests FAILED  — full log: ${LOG_FILE}"

    exit 1

fi

src/tests/ecoc26_pluggables/run_in_K8s/zsm_test_job.yaml

0 → 100644
+75 −0
Original line number Diff line number Diff line 
# Copyright 2022-2026 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

#

# Kubernetes Job: ZSM failure notification test

#

# Runs src/tests/ecoc26_pluggables/tests/test_zsm_failure_notification.py inside

# the tfs namespace so that Kubernetes service env vars are automatically

# injected for context, device, kpi-manager, telemetry and automation services.

#

# Prerequisites:

#   - The tfs namespace must be deployed (deploy/tfs.sh).

#   - The host path /home/cttc/tfs-ctrl must be accessible on every K8s node

#     (true for single-node MicroK8s).

#

# Usage: see run_zsm_test.sh



apiVersion: batch/v1

kind: Job

metadata:

  name: zsm-automation-test

  namespace: tfs

spec:

  ttlSecondsAfterFinished: 600   # auto-delete the Job 10 min after it finishes

  backoffLimit: 0                # do not retry on failure

  template:

    spec:

      restartPolicy: Never

      containers:

        - name: test-runner

          # automation image: has pytest, all compiled proto stubs under

          # /var/teraflow/common/proto/, and the AutomationClient.

          # kpi_manager, context, and device clients are loaded from the

          # hostPath-mounted source tree via PYTHONPATH entry 2.

          image: localhost:32000/tfs/automation:dev

          imagePullPolicy: IfNotPresent

          # Run from the project root so that "src." package imports resolve.

          workingDir: /home/cttc/tfs-ctrl

          command:

            - python

            - -m

            - pytest

            - --log-level=DEBUG

            - --log-cli-level=DEBUG

            - --verbose

            - src/tests/ecoc26_pluggables/tests/test_zsm_failure_notification.py

          env:

            # PYTHONPATH resolution order:

            #   1. /var/teraflow              → compiled common.proto stubs,

            #                                   automation.*, common.*

            #   2. /home/cttc/tfs-ctrl/src    → kpi_manager.client,

            #                                   context.client, device.client,

            #                                   automation.client, etc.

            #   3. /home/cttc/tfs-ctrl        → src.tests.ecoc26_pluggables.*

            - name: PYTHONPATH

              value: "/var/teraflow:/home/cttc/tfs-ctrl/src:/home/cttc/tfs-ctrl"

          volumeMounts:

            - name: tfs-source

              mountPath: /home/cttc/tfs-ctrl

              readOnly: true

      volumes:

        - name: tfs-source

          hostPath:

            path: /home/cttc/tfs-ctrl

            type: Directory