issue: 4197123: Add Pytest to the workflow of the LogAnalyzer + issue: 4197133: Add Pytest

.github/workflows/ufm_log_analyzer_ci_workflow.yml

@@ -53,3 +53,29 @@ jobs:
         pip install ruff==0.7.3
 
         ruff format --diff --check src/loganalyze
+
+  pytest:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: 3.9
+
+    - name: Run Pytest
+      run: |
+        SCRIPT_DIR="plugins/ufm_log_analyzer_plugin"
+        # Set PYTHONPATH to include src directory and two levels up for utils
+        PYTHONPATH="$(realpath $SCRIPT_DIR/src):$(realpath $SCRIPT_DIR/../../)"
+        export PYTHONPATH
+
+        cd $SCRIPT_DIR
+
+        pip install -r src/loganalyze/requirements.txt
+        pip install pytest
+
+        pytest --maxfail=1 --disable-warnings -q  # Runs pytest with limited output

plugins/ufm_log_analyzer_plugin/src/loganalyze/log_analyzers/base_analyzer.py

@@ -185,7 +185,11 @@ def __init__(
     ):
         super().__init__(dest_image_path)
         dataframes = [pd.read_csv(ufm_log) for ufm_log in logs_csvs]
-        df = pd.concat(dataframes, ignore_index=True)
+        if dataframes:
+            df = pd.concat(dataframes, ignore_index=True)
+        else:
+            df = pd.DataFrame()  # Return an empty DataFrame if dataframes is empty
+
         if sort_timestamp:
             df[DataConstants.TIMESTAMP] = pd.to_datetime(df[DataConstants.TIMESTAMP])
             max_timestamp = df[DataConstants.TIMESTAMP].max()

plugins/ufm_log_analyzer_plugin/unit_tests/test_ibdiagnet_log_analyzer.py

@@ -0,0 +1,46 @@
+"""
+@copyright:
+    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.
+
+@author: Miryam Schwartz
+@date:   Dec 08, 2024
+"""
+
+import unittest
+import sys
+import os
+
+sys.path.append(os.getcwd())
+sys.path.append("/".join(os.getcwd().split("/")[:-1]))
+sys.path.append("/".join(os.getcwd().split("/")[:-1]) + "/src")
+
+from loganalyze.log_analyzers.ibdiagnet_log_analyzer import IBDIAGNETLogAnalyzer
+
+
+class TestIBDIAGNETLogAnalyzer(unittest.TestCase):
+    def setUp(self):
+        # Example initialization with dummy arguments
+        self.logs_csvs = []
+        self.hours = 24
+        self.dest_image_path = "/dummy/path"
+        self.analyzer = IBDIAGNETLogAnalyzer(self.logs_csvs, self.hours, self.dest_image_path)
+
+    def test_get_fabric_size(self):
+        # Mock the _log_data_sorted attribute
+        expected_fabric_size = {"switch_count": 10, "link_count": 50}  # Example data
+        self.analyzer._log_data_sorted = expected_fabric_size # pylint: disable=protected-access
+
+        # Call the method and check the result
+        result = self.analyzer.get_fabric_size()
+        self.assertEqual(result, expected_fabric_size)
+
+if __name__ == "__main__":
+    unittest.main()

plugins/ufm_log_analyzer_plugin/unit_tests/test_ufm_top_analyzer.py

@@ -0,0 +1,49 @@
+"""
+@copyright:
+    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.
+
+@author: Miryam Schwartz
+@date:   Dec 08, 2024
+"""
+
+import unittest
+import sys
+import os
+
+sys.path.append(os.getcwd())
+sys.path.append("/".join(os.getcwd().split("/")[:-1]))
+sys.path.append("/".join(os.getcwd().split("/")[:-1]) + "/src")
+
+from loganalyze.log_analyzers.ufm_top_analyzer import UFMTopAnalyzer
+
+class TestUFMTopAnalyzer(unittest.TestCase):
+    def setUp(self):
+        self.analyzer = UFMTopAnalyzer()
+
+    def test_add_analyzer(self):
+        mock_analyzer_1 = "Analyzer1"
+        mock_analyzer_2 = "Analyzer2"
+
+        # Initially, the list should be empty
+        self.assertEqual(len(self.analyzer._analyzers), 0) # pylint: disable=protected-access
+
+        # Add first analyzer and check the length
+        self.analyzer.add_analyzer(mock_analyzer_1)
+        self.assertEqual(len(self.analyzer._analyzers), 1) # pylint: disable=protected-access
+        self.assertIn(mock_analyzer_1, self.analyzer._analyzers) # pylint: disable=protected-access
+
+        # Add second analyzer and check the updated length
+        self.analyzer.add_analyzer(mock_analyzer_2)
+        self.assertEqual(len(self.analyzer._analyzers), 2) # pylint: disable=protected-access
+        self.assertIn(mock_analyzer_2, self.analyzer._analyzers) # pylint: disable=protected-access
+
+if __name__ == "__main__":
+    unittest.main()