Issue:3897880:PDR exclusion list tests (#217)

.github/workflows/pdr_plugin_ci_workflow.yml

@@ -0,0 +1,51 @@
+name: PDR Plugin CI Workflow
+
+on:
+  push:
+     paths:
+      - 'plugins/pdr_deterministic_plugin/**'
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    env:
+      PYTHONPATH: '.:plugins/pdr_deterministic_plugin/ufm_sim_web_service'
+      PDRPATH: 'plugins/pdr_deterministic_plugin'
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@main
+
+    - name: Set up Python
+      uses: actions/setup-python@main
+      with:
+        python-version: 3.9
+
+    - name: Install dependencies
+      run: |
+        pip install -r $PDRPATH/requirements.txt
+        pip install pylint
+        pip install pytest-cov
+
+    - name: Run PyLint
+      run: pylint --rcfile=$PDRPATH/.pylintrc $PDRPATH
+
+    - name: Run exclusion list class test
+      timeout-minutes: 5
+      run: pytest -s $PDRPATH/tests/exclude_list_class_tests.py --cov=$PDRPATH
+
+    - name: Test exclusion list REST API
+      timeout-minutes: 5
+      run: |
+        sudo bash $PDRPATH/.pytest/run_pdr_standalone_pytest.sh
+        echo "Test exclusion list REST API methods"
+        pytest -s $PDRPATH/tests/exclude_list_rest_api_tests.py --cov=$PDRPATH
+        sudo bash $PDRPATH/.pytest/terminate_pdr_standalone_pytest.sh
+
+    - name: Run full simulation test
+      timeout-minutes: 10
+      run: |
+        sudo bash $PDRPATH/.pytest/run_pdr_standalone_pytest.sh
+        echo "Starting simulated test"
+        python $PDRPATH/tests/simulation_telemetry.py
+        sudo bash $PDRPATH/.pytest/terminate_pdr_standalone_pytest.sh

plugins/pdr_deterministic_plugin/.pylintrc

@@ -0,0 +1,17 @@
+[MASTER]
+init-hook="import os, sys; sys.path.append(os.path.join(os.getcwd(), 'plugins', 'pdr_deterministic_plugin', 'src')); sys.path.append(os.path.join(os.getcwd(), 'utils'))"
+
+[MAIN]
+max-public-methods=100
+
+[DESIGN]
+max-attributes=10
+
+[MESSAGES CONTROL]
+disable=missing-module-docstring,missing-function-docstring,fixme
+
+[FORMAT]
+max-line-length=140
+
+[BASIC]
+min-public-methods=0

plugins/pdr_deterministic_plugin/.pytest/run_pdr_standalone_pytest.sh

@@ -0,0 +1,39 @@
+#!/bin/bash -x
+
+PLUGIN_DIR="plugins/pdr_deterministic_plugin"
+pip install -r $PLUGIN_DIR/requirements.txt >/dev/null 2>&1
+
+cp -r utils $PLUGIN_DIR/ufm_sim_web_service
+cp -r utils $PLUGIN_DIR/tests
+
+echo "Init PDR configuration file"
+CONFIG_FILE="/config/pdr_deterministic.conf"
+mkdir -p /config
+cp -f $PLUGIN_DIR/build/config/pdr_deterministic.conf "$CONFIG_FILE"
+sed -i -e 's/DRY_RUN=False/DRY_RUN=True/g' "$CONFIG_FILE"
+sed -i -e 's/INTERVAL=300/INTERVAL=10/g' "$CONFIG_FILE"
+sed -i -e 's/CONFIGURED_TEMP_CHECK=False/CONFIGURED_TEMP_CHECK=True/g' "$CONFIG_FILE"
+sed -i -e 's/LINK_DOWN_ISOLATION=False/LINK_DOWN_ISOLATION=True/g' "$CONFIG_FILE"
+sed -i -e 's/DEISOLATE_CONSIDER_TIME=5/DEISOLATE_CONSIDER_TIME=1/g' "$CONFIG_FILE"
+
+# Remove any existing TEST_MODE lines from the file
+sed -i -e '/TEST_MODE=\(True\|False\)/d' "$CONFIG_FILE"
+
+# Check if the section [Common] exists
+if grep -q '^\[Common\]' "$CONFIG_FILE"; then
+    # If section exists, insert TEST_MODE=True under it
+    awk '/^\[Common\]/ {print; print "TEST_MODE=True"; next}1' "$CONFIG_FILE" > temp && mv temp "$CONFIG_FILE"
+else
+    # If section does not exist, add it to the file
+    echo -e '\n[Common]\nTEST_MODE=True\n' >> "$CONFIG_FILE"
+fi
+
+bash $PLUGIN_DIR/.pytest/terminate_pdr_standalone_pytest.sh
+echo "Starting standalone PDR process"
+python $PLUGIN_DIR/ufm_sim_web_service/isolation_algo.py >/dev/null 2>&1 &
+sleep 10
+if ! ps aux | grep -q [i]solation_algo.py; then
+    echo "Failed to start standalone PDR process"
+    exit 1
+fi
+# PDR is up and ready for communication

plugins/pdr_deterministic_plugin/.pytest/terminate_pdr_standalone_pytest.sh

@@ -0,0 +1,9 @@
+#!/bin/bash -x
+
+echo "Terminating standalone PDR process"
+ps -ef | grep isolation_algo.py | grep -v grep | awk '{print $2}' | xargs kill -9 $1 >/dev/null 2>/dev/null
+sleep 10
+if ps aux | grep -q [i]solation_algo.py; then
+    echo "Failed to terminate standalone PDR process"
+    exit 1
+fi

plugins/pdr_deterministic_plugin/__init__.py

@@ -0,0 +1,11 @@
+#
+# Copyright © 2013-2024 NVIDIA CORPORATION & AFFILIATES. ALL RIGHTS RESERVED.
+#
+# This software product is a proprietary product of Nvidia Corporation and its affiliates
+# (the "Company") and all right, title, and interest in and to the software
+# product, including all associated intellectual property rights, are and
+# shall remain exclusively with the Company.
+#
+# This software product is governed by the End User License Agreement
+# provided with the software product.
+#

plugins/pdr_deterministic_plugin/requirements.txt

@@ -0,0 +1,9 @@
+flask<=3.0.3
+numpy<=1.26.4
+pandas<=2.2.2
+pytest<=8.2.0
+requests<=2.31.0
+twisted<=22.1.0
+flask_restful<=0.3.10
+tzlocal<=4.2
+jsonschema<=4.5.1

plugins/pdr_deterministic_plugin/tests/exclude_list_class_tests.py

@@ -0,0 +1,102 @@
+#
+# Copyright © 2013-2024 NVIDIA CORPORATION & AFFILIATES. ALL RIGHTS RESERVED.
+#
+# This software product is a proprietary product of Nvidia Corporation and its affiliates
+# (the "Company") and all right, title, and interest in and to the software
+# product, including all associated intellectual property rights, are and
+# shall remain exclusively with the Company.
+#
+# This software product is governed by the End User License Agreement
+# provided with the software product.
+#
+
+import os
+import tempfile
+import time
+from constants import PDRConstants as Constants
+from exclude_list import ExcludeList, ExcludeListItem
+from isolation_algo import create_logger
+
+
+def get_logger():
+    """
+    Return logger associated with log file in temporary directory
+    """
+    log_name = os.path.basename(Constants.LOG_FILE)
+    log_path = os.path.join(tempfile.gettempdir(), log_name)
+    return create_logger(log_path)
+
+
+def test_exclude_list_class_methods():
+    """
+    Test exclude list class methods by direct calls
+    """
+    print("\n") # Start tests output from new line
+
+    excluded_ports = [
+        ExcludeListItem("0123456789aaabbb_1", 0),  # Add forever
+        ExcludeListItem("9876543210cccddd_2", 30), # Add for 30 seconds
+        ExcludeListItem("3456789012eeefff_3", 0)   # Add forever
+    ]
+
+    # Create exclusion list and ensure it's empty
+    exclude_list = ExcludeList(get_logger())
+    items = exclude_list.items()
+    assert not items
+    print("    - test: create exclusion list and ensure it's empty -- PASS")
+
+    # Add ports to excluded list
+    for port in excluded_ports:
+        exclude_list.add(port.port_name, port.ttl_seconds)
+
+    # Test exclusion list size
+    items = exclude_list.items()
+    assert items and len(items) == len(excluded_ports)
+    print("    - test: add ports to exclusion list -- PASS")
+
+    # Test 'contains' method
+    for port in excluded_ports:
+        assert exclude_list.contains(port.port_name)
+    print("    - test: exclusion list 'contains' method -- PASS")
+
+    # Test exclusion list content
+    for (index, item) in enumerate(items):
+        assert item.port_name == excluded_ports[index].port_name
+        assert item.ttl_seconds == excluded_ports[index].ttl_seconds
+    print("    - test: exclusion list content -- PASS")
+
+    # Test auto-remove of second port after TTL is expired
+    auto_remove_port = excluded_ports[1]
+    time.sleep(auto_remove_port.ttl_seconds + 1)
+    assert not exclude_list.contains(auto_remove_port.port_name)
+    print("    - test: auto-remove of port from exclusion list after TTL is expired -- PASS")
+
+    # Test excluded list size
+    items = exclude_list.items()
+    assert items and len(items) == (len(excluded_ports) - 1)
+
+    # Test excluded list content
+    for port in excluded_ports:
+        if port.port_name != auto_remove_port.port_name:
+            assert exclude_list.contains(port.port_name)
+    print("    - test: exclusion list content -- PASS")
+
+    # Test forced remove of third port
+    remove_port = excluded_ports[2]
+    exclude_list.remove(port.port_name)
+    assert not exclude_list.contains(remove_port.port_name)
+    print("    - test: forced remove of port from exclusion list -- PASS")
+
+    # Test excluded list size
+    items = exclude_list.items()
+    assert items and len(items) == (len(excluded_ports) - 2)
+
+    # Test excluded list content
+    for port in excluded_ports:
+        if port.port_name != remove_port.port_name and port.port_name != auto_remove_port.port_name:
+            assert exclude_list.contains(port.port_name)
+    print("    - test: exclusion list content -- PASS")
+
+
+if __name__ == '__main__':
+    test_exclude_list_class_methods()

plugins/pdr_deterministic_plugin/tests/exclude_list_rest_api_tests.py

@@ -0,0 +1,99 @@
+#
+# Copyright © 2013-2024 NVIDIA CORPORATION & AFFILIATES. ALL RIGHTS RESERVED.
+#
+# This software product is a proprietary product of Nvidia Corporation and its affiliates
+# (the "Company") and all right, title, and interest in and to the software
+# product, including all associated intellectual property rights, are and
+# shall remain exclusively with the Company.
+#
+# This software product is governed by the End User License Agreement
+# provided with the software product.
+#
+
+import http
+import json
+import random
+import time
+import requests
+
+
+def generate_port_name():
+    """
+    Generate port name
+    """
+    port_guid = f'{random.randrange(16**16):016x}'
+    port_num = random.randint(10, 99)
+    return f'{port_guid}_{port_num}'
+
+
+def test_exclude_list_rest_api():
+    """
+    Test exclude list via plugin REST API
+    """
+    print("\n") # Start tests output from new line
+
+    url = "http://127.0.0.1:8977/excluded"
+
+    excluded_ports = [
+        (generate_port_name(), 0),  # Add forever
+        (generate_port_name(), 30), # Add for 30 seconds
+        (generate_port_name(), 0)   # Add forever
+    ]
+
+    # Get (empty) list content
+    response = requests.get(url, timeout=5)
+    assert response.status_code == http.client.OK
+    assert all(char.isspace() for char in response.text)
+    print("    - test: get exclusion list and ensure it's empty -- PASS")
+
+    # Add ports to excluded list
+    response = requests.put(url, data=json.dumps(excluded_ports), timeout=5)
+    assert response.status_code == http.client.OK
+    for pair in excluded_ports:
+        port_name = pair[0]
+        assert port_name in response.text
+    print("    - test: add ports to exclusion list -- PASS")
+
+    # Test exclusion list content
+    response = requests.get(url, timeout=5)
+    assert response.status_code == http.client.OK
+    for pair in excluded_ports:
+        port_name = pair[0]
+        assert port_name in response.text
+    print("    - test: get added ports from exclusion list -- PASS")
+
+    # Wait until second port TTL is expired
+    ttl_seconds = excluded_ports[1][1]
+    time.sleep(ttl_seconds + 1)
+
+    # Test auto-remove of second port after TTL is expired
+    response = requests.get(url, timeout=5)
+    assert response.status_code == http.client.OK
+    for (index, pair) in enumerate(excluded_ports):
+        port_name = pair[0]
+        if index == 1:
+            assert port_name not in response.text
+        else:
+            assert port_name in response.text
+    print("    - test: auto-remove of port from exclusion list after TTL is expired -- PASS")
+
+    # Test forced remove of third port
+    port_name = excluded_ports[2][0]
+    response = requests.delete(url, data=json.dumps([port_name]), timeout=5)
+    assert response.status_code == http.client.OK
+    assert f'{port_name} removed' in response.text
+    print("    - test: forced remove of port from exclusion list -- PASS")
+
+    # Test exclusion list content
+    response = requests.get(url, timeout=5)
+    for (index, pair) in enumerate(excluded_ports):
+        port_name = excluded_ports[index][0]
+        if index == 1 or index == 2:
+            assert port_name not in response.text
+        else:
+            assert port_name in response.text
+    print("    - test: exclusion list content -- PASS")
+
+
+if __name__ == '__main__':
+    test_exclude_list_rest_api()