Updated MWC 2026 SIMAP implementation

# Run the test with logging enabled and capture output

# Run pytest for the test_analyze_endpoint test case with verbose and debug logging.

# 

# Arguments:

#   $TEST_FILE - Path to the test file containing test_analyze_endpoint

#   $LOG_FILE - Path where debug logs will be written

#   "$@" - Forwards all additional command-line arguments passed to this script

#        to pytest (e.g., additional pytest options, markers, or flags)

#

# Options:

#   -v: Verbose output

#   -s: Disable output capturing (show print statements)

#   --log-cli-level=DEBUG: Set console logging level to DEBUG

#   --log-file: Specify log file path

#   --log-file-level=DEBUG: Set file logging level to DEBUG

pytest $TEST_FILE::test_analyze_endpoint \

    -v -s \

    --log-cli-level=DEBUG \

    --log-file="${LOG_FILE}" \

    --log-file-level=DEBUG \

    "$@"

echo "Running test_notify_endpoint_upgrade_downgrade..."

pytest $TEST_FILE::test_notify_endpoint_upgrade_downgrade \

    -v -s \

    --log-cli-level=DEBUG \

    --log-file="${LOG_FILE}" \

    --log-file-level=DEBUG \

    "$@"

echo "Running test_stop_all_analyses_endpoint..."

pytest $TEST_FILE::test_stop_all_analyses_endpoint \

# pytest $TEST_FILE::test_analyze_endpoint \

    -v -s \

    --log-cli-level=DEBUG \

    --log-file="${LOG_FILE}" \

    --log-file-level=DEBUG \

    "$@"

echo ""

echo "Test logs saved to: ${LOG_FILE}"

        "history_window_size_sec":      60,

        "forecast_sample_interval_sec": 30,

        "forecast_sample_count":        50,

        "duration_minutes":             10  # Short duration for testing

        "duration_minutes":             30,

    }

    LOGGER.info(f"Sending analyze request with payload: {payload}")

        LOGGER.info("Analysis started successfully")

        assert data['status']             ==  'accepted',       f"Expected status 'accepted',         got '{data['status']}'"

        assert data['simap_id']           ==  'E2E-L1',         f"Expected simap_id 'E2E-L1',         got '{data['simap_id']}'"

        assert data['duration_minutes']   ==  2,                f"Expected duration_minutes 2,        got '{data['duration_minutes']}'"

        assert data['duration_minutes']   ==  30,               f"Expected duration_minutes 30,       got '{data['duration_minutes']}'"

        assert '/osm/aiAnalyticsEvent/v1' in  data['endpoint'], f"Expected '/osm/aiAnalyticsEvent/v1' in  endpoint"

    elif response.status_code == 503:

        assert data['status'] == 'error', f"Expected status 'error' for 503, got '{data['status']}'"

    LOGGER.info("<<<<<< Finished test_case test_analyze_endpoint")

def test_status_endpoint(ai_engine_server):

def test_status_endpoint(ai_engine_server_connection_confirmation):

    """

    Test GET /api/v1/status endpoint.

    LOGGER.info("<<<<<< Finished test_case test_status_endpoint")

def test_stop_analyze_endpoint(ai_engine_server):

def test_stop_analyze_endpoint(ai_engine_server_connection_confirmation):

    """

    Test POST /api/v1/analyze/stop endpoint.

    LOGGER.info("<<<<<< Finished test_case test_stop_analyze_endpoint")

def test_stop_all_analyses_endpoint(ai_engine_server):

def test_stop_all_analyses_endpoint(ai_engine_server_connection_confirmation):

    """

    Test POST /api/v1/analyze/stop-all endpoint.

    LOGGER.info(">>>>>> Starting test_case test_stop_all_analyses_endpoint: POST /api/v1/analyze/stop-all endpoint")

    started_ids = ["E2E-L1"]

    LOGGER.info("Waiting 120 seconds before starting analyses to test stop-all functionality...")

    time.sleep(120)

    # Only proceed if at least one analysis started

    if len(started_ids) > 0:

    LOGGER.info("<<<<<< Finished test_case test_stop_all_analyses_endpoint")

# TODO: Add here test for notify endpoint from     @blueprint.route('/notify', methods=['POST'])

 No newline at end of file

def test_notify_endpoint_upgrade_downgrade(ai_engine_server_connection_confirmation):

    """

    Test POST /api/v1/notify endpoint with upgrade and downgrade notifications.

    Validates that the notify endpoint:

    - Accepts upgrade notification

    - Waits 30 seconds

    - Accepts downgrade notification

    - Returns appropriate responses for both

    """

    LOGGER.info(">>>>>> Starting test_case test_notify_endpoint_upgrade_downgrade: POST /api/v1/notify endpoint")

    LOGGER.info("waiting 20 seconds before sending UPGRADE notification...")

    time.sleep(20)

    # Prepare upgrade notification payload (API expects 'status' key only)

    upgrade_payload = {

        "status": "UPGRADE"

    }

    LOGGER.info(f"Sending UPGRADE notification with payload: {upgrade_payload}")

    # Send POST request for upgrade notification

    upgrade_response = requests.post(

        f'{BASE_URL}/api/v1/notify',

        json=upgrade_payload,

        timeout=10

    )

    LOGGER.info(f"Upgrade notification response status: {upgrade_response.status_code}")

    # Parse JSON response

    upgrade_data = upgrade_response.json()

    LOGGER.info(f"Upgrade notification response body: {upgrade_data}")

    # Validate upgrade response

    assert 'status' in upgrade_data, "Upgrade response missing 'status' field"

    # Accept either success (200) or accepted (202)

    if upgrade_response.status_code in [200, 202]:

        LOGGER.info("Upgrade notification sent successfully")

    elif upgrade_response.status_code == 503:

        LOGGER.warning("Service unavailable for upgrade notification")

        pytest.skip("External service unavailable (expected if SIMAP/InfluxDB not running)")

    else:

        LOGGER.error(f"Unexpected status code for upgrade: {upgrade_response.status_code}")

        pytest.fail(f"Unexpected status code: {upgrade_response.status_code}")

    # Wait 60 seconds before sending downgrade

    LOGGER.info("Waiting 60 seconds before sending DOWNGRADE notification...")

    time.sleep(60)

    # Prepare downgrade notification payload (API expects 'status' key only)

    downgrade_payload = {

        "status": "DOWNGRADE"

    }

    LOGGER.info(f"Sending DOWNGRADE notification with payload: {downgrade_payload}")

    # Send POST request for downgrade notification

    downgrade_response = requests.post(

        f'{BASE_URL}/api/v1/notify',

        json=downgrade_payload,

        timeout=10

    )

    LOGGER.info(f"Downgrade notification response status: {downgrade_response.status_code}")

    # Parse JSON response

    downgrade_data = downgrade_response.json()

    LOGGER.info(f"Downgrade notification response body: {downgrade_data}")

    # Validate downgrade response

    assert 'status' in downgrade_data, "Downgrade response missing 'status' field"

    # Accept either success (200) or accepted (202)

    if downgrade_response.status_code in [200, 202]:

        LOGGER.info("Downgrade notification sent successfully")

    elif downgrade_response.status_code == 503:

        LOGGER.warning("Service unavailable for downgrade notification")

        pytest.skip("External service unavailable (expected if SIMAP/InfluxDB not running)")

    else:

        LOGGER.error(f"Unexpected status code for downgrade: {downgrade_response.status_code}")

        pytest.fail(f"Unexpected status code: {downgrade_response.status_code}")

    LOGGER.info("Notify endpoint upgrade/downgrade test passed!")

    LOGGER.info("<<<<<< Finished test_case test_notify_endpoint_upgrade_downgrade")

 No newline at end of file

    Access links are more sensitive to load than core links.

    """

    def __init__(self, service_count: int = 0):

    def __init__(self, service_count: int = 0, seed: int = None):

        LOGGER.info("Initiating SimapMetricsGenerator")

        self._random = random.Random(seed)

        self._service_count = 0

        self._service_ids: Dict[str, List[str]] = {

            'te'    : [],

        latency        = base_latency * (1.0 + congestion_factor * sensitivity * 4.0)

        # Add uniform noise (5%)

        bw_noise  = random.uniform(-0.05, 0.05) * bw_utilization

        lat_noise = random.uniform(-0.05, 0.05) * latency

        bw_noise  = self._random.uniform(-0.05, 0.05) * bw_utilization

        lat_noise = self._random.uniform(-0.05, 0.05) * latency

        bw_utilization = max(0.0, min(100.0, bw_utilization + bw_noise))

        latency        = max(0.1, latency + lat_noise)

# RESTCONF/SIMAP Server

This server implements a basic RESTCONF Server that can load, potentially, any YANG data model.

In this case, it is prepared to load a SIMAP Server based on IETF Network Topology + custom SIMAP Telemetry extensions.

## Build the RESTCONF/SIMAP Server Docker image

```bash

./build.sh

```

## Deploy the RESTCONF/SIMAP Server

```bash

./deploy.sh

```

## Run the RESTCONF/SIMAP Client for testing:

```bash

./run_client.sh

```

The telemetry client is schedule-driven. Accepted runs must record a run ID,

integer seed, named profile, duration, sample interval, and JSON Lines event

log. For example:

```bash

./run_scheduled_client.sh --run-id smoke-step-seed-7 --seed 7 --profile step \

  --duration 900 --sample-interval 10 \

  --event-log /path/to/run/simap-source.events.jsonl

```

Profiles are `step`, `ramp`, `burst`, `periodic`,

`legacy_connection_count`, and `custom`. The legacy profile precomputes its

old connection-count changes with the recorded seed; it is retained only for

comparability. A custom schedule is JSON with `profile: "custom"` and a

strictly increasing `points` list containing `at_seconds` and

`connection_count` values. Load is determined only by elapsed time and never

by a BE/DET policy action.

`run_client.sh` and `simap_client/__main__.py` retain the legacy behaviour for

backward compatibility. The new schedule-driven implementation is isolated in

`simap_client/scheduled_main.py` and is started only through

`run_scheduled_client.sh`.

## Destroy the RESTCONF/SIMAP Server

```bash

./destroy.sh

```

#!/bin/bash

# Copyright 2022-2025 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.

# Make folder containing the script the root folder for its execution.

cd "$(dirname "$0")/../../../" || exit 1

python3 -m tests.tools.simap_server.simap_client.scheduled_main "$@"