Add Python UTs (#2186)

* Fix duplicate config filenames and add unit tests with CI

- Fix telegraf config filename collision: include plugin name in filename
  to prevent plugins in the same omiclass from overwriting each other
  (e.g. mem/swap/kernel_vmstat under 'memory')
- Add comprehensive unit tests for all LAD-AMA-Common modules (91 tests)
- Add GitHub Actions CI workflow triggered on LAD-AMA-Common changes

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Add unit tests for AzureMonitorAgent with CI and packaging exclusion

- Add 104 unit tests covering agent.py, error handling, helpers, OS
  checks, and logrotate validation
- Add GitHub Actions CI workflow for AzureMonitorAgent
- Exclude tests/ directory from packaging zip in packaging.sh

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: simathih

.github/workflows/azure-monitor-agent-tests.yml

@@ -0,0 +1,34 @@
+name: AzureMonitorAgent Tests
+
+on:
+  push:
+    branches: [master, main]
+    paths:
+      - 'AzureMonitorAgent/**'
+      - 'LAD-AMA-Common/**'
+  pull_request:
+    branches: [master, main]
+    paths:
+      - 'AzureMonitorAgent/**'
+      - 'LAD-AMA-Common/**'
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.9', '3.10', '3.11', '3.12']
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Run AzureMonitorAgent tests
+        working-directory: AzureMonitorAgent
+        env:
+          PYTHONPATH: ${{ github.workspace }}/LAD-AMA-Common
+        run: python -m unittest discover -s tests -v

AzureMonitorAgent/packaging.sh

@@ -96,7 +96,7 @@ fi
 
 echo "Packaging extension $PACKAGE_NAME to $output_path"
 excluded_files="agent.version packaging.sh apply_version.sh update_version.sh"
-zip -r $output_path/$PACKAGE_NAME * -x $excluded_files "./test/*" "./extension-test/*" "./references" "./tmp"
+zip -r $output_path/$PACKAGE_NAME * -x $excluded_files "./test/*" "./tests/*" "./extension-test/*" "./references" "./tmp"
 
 # validate package size is within limits; these limits come from arc, ideally they are removed in the future
 max_uncompressed_size=$((1000 * 1024 * 1024))

AzureMonitorAgent/tests/__init__.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python
+"""
+Unit tests for AzureMonitorAgent/agent.py - pure logic functions only.
+"""
+
+import sys
+import os
+import re
+import unittest
+from unittest.mock import patch, MagicMock
+
+# Mock Linux-only modules before importing
+for mod_name in ('grp', 'pwd'):
+    if mod_name not in sys.modules:
+        sys.modules[mod_name] = MagicMock()
+
+# Mock waagent/HUtil imports that agent.py tries to load at module level
+sys.modules['Utils'] = MagicMock()
+sys.modules['Utils.WAAgentUtil'] = MagicMock()
+sys.modules['Utils.WAAgentUtil'].waagent = MagicMock()
+sys.modules['Utils.HandlerUtil'] = MagicMock()
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'LAD-AMA-Common')))
+
+import agent
+
+
+class TestIsTruthy(unittest.TestCase):
+    """Tests for is_truthy function."""
+
+    def test_true_bool(self):
+        self.assertTrue(agent.is_truthy(True))
+
+    def test_false_bool(self):
+        self.assertFalse(agent.is_truthy(False))
+
+    def test_true_string(self):
+        self.assertTrue(agent.is_truthy("true"))
+
+    def test_true_string_capitalized(self):
+        self.assertTrue(agent.is_truthy("True"))
+
+    def test_true_string_uppercase(self):
+        self.assertTrue(agent.is_truthy("TRUE"))
+
+    def test_true_string_with_whitespace(self):
+        self.assertTrue(agent.is_truthy("  true  "))
+
+    def test_false_string(self):
+        self.assertFalse(agent.is_truthy("false"))
+
+    def test_none(self):
+        self.assertFalse(agent.is_truthy(None))
+
+    def test_empty_string(self):
+        self.assertFalse(agent.is_truthy(""))
+
+    def test_zero(self):
+        self.assertFalse(agent.is_truthy(0))
+
+    def test_one(self):
+        self.assertFalse(agent.is_truthy(1))
+
+
+class TestIsDpkgOrRpmLocked(unittest.TestCase):
+    """Tests for is_dpkg_or_rpm_locked."""
+
+    def test_dpkg_locked(self):
+        output = "E: Could not get lock /var/lib/dpkg/lock - open"
+        self.assertTrue(agent.is_dpkg_or_rpm_locked(1, output))
+
+    def test_rpm_locked(self):
+        output = "error: waiting for transaction rpm lock on /var/lib/rpm/.rpm.lock"
+        self.assertTrue(agent.is_dpkg_or_rpm_locked(1, output))
+
+    def test_not_locked_success(self):
+        self.assertFalse(agent.is_dpkg_or_rpm_locked(0, "Success"))
+
+    def test_not_locked_other_error(self):
+        self.assertFalse(agent.is_dpkg_or_rpm_locked(1, "Package not found"))
+
+
+class TestRetryIfDpkgOrRpmLocked(unittest.TestCase):
+    """Tests for retry_if_dpkg_or_rpm_locked."""
+
+    def test_locked_returns_retry(self):
+        result = agent.retry_if_dpkg_or_rpm_locked(
+            1, "dpkg status database is locked by another process"
+        )
+        self.assertTrue(result[0])
+
+    def test_not_locked_no_retry(self):
+        result = agent.retry_if_dpkg_or_rpm_locked(0, "ok")
+        self.assertFalse(result[0])
+
+
+class TestFinalCheckIfDpkgOrRpmLocked(unittest.TestCase):
+    """Tests for final_check_if_dpkg_or_rpm_locked."""
+
+    def test_locked_returns_special_code(self):
+        code = agent.final_check_if_dpkg_or_rpm_locked(
+            1, "dpkg lock held"
+        )
+        self.assertEqual(code, agent.DPKGOrRPMLockedErrorCode)
+
+    def test_not_locked_returns_original_code(self):
+        code = agent.final_check_if_dpkg_or_rpm_locked(0, "ok")
+        self.assertEqual(code, 0)
+
+
+class TestValidatePortNumber(unittest.TestCase):
+    """Tests for validate_port_number."""
+
+    @patch('agent.hutil_log_error')
+    def test_valid_port(self, mock_log):
+        self.assertEqual(agent.validate_port_number("8080", "test"), "8080")
+
+    @patch('agent.hutil_log_error')
+    def test_valid_port_with_whitespace(self, mock_log):
+        self.assertEqual(agent.validate_port_number("  443  ", "test"), "443")
+
+    @patch('agent.hutil_log_error')
+    def test_port_min(self, mock_log):
+        self.assertEqual(agent.validate_port_number("1", "test"), "1")
+
+    @patch('agent.hutil_log_error')
+    def test_port_max(self, mock_log):
+        self.assertEqual(agent.validate_port_number("65535", "test"), "65535")
+
+    @patch('agent.hutil_log_error')
+    def test_port_zero_invalid(self, mock_log):
+        self.assertEqual(agent.validate_port_number("0", "test"), "")
+
+    @patch('agent.hutil_log_error')
+    def test_port_too_high(self, mock_log):
+        self.assertEqual(agent.validate_port_number("65536", "test"), "")
+
+    @patch('agent.hutil_log_error')
+    def test_port_negative(self, mock_log):
+        self.assertEqual(agent.validate_port_number("-1", "test"), "")
+
+    @patch('agent.hutil_log_error')
+    def test_port_non_numeric(self, mock_log):
+        self.assertEqual(agent.validate_port_number("abc", "test"), "")
+
+    @patch('agent.hutil_log_error')
+    def test_port_empty(self, mock_log):
+        self.assertEqual(agent.validate_port_number("", "test"), "")
+
+    @patch('agent.hutil_log_error')
+    def test_port_none(self, mock_log):
+        self.assertEqual(agent.validate_port_number(None, "test"), "")
+
+
+class TestGetProxyMode(unittest.TestCase):
+    """Tests for get_proxy_mode."""
+
+    def test_none_settings(self):
+        self.assertIsNone(agent.get_proxy_mode(None))
+
+    def test_no_proxy_key(self):
+        self.assertIsNone(agent.get_proxy_mode({"other": "value"}))
+
+    def test_proxy_no_mode(self):
+        self.assertIsNone(agent.get_proxy_mode({"proxy": {}}))
+
+    def test_proxy_with_mode(self):
+        self.assertEqual(
+            agent.get_proxy_mode({"proxy": {"mode": "application"}}),
+            "application"
+        )
+
+    def test_proxy_none_value(self):
+        self.assertIsNone(agent.get_proxy_mode({"proxy": None}))
+
+
+class TestGetServiceCommand(unittest.TestCase):
+    """Tests for get_service_command."""
+
+    @patch('agent.is_systemd', return_value=True)
+    def test_systemd_single_op(self, _):
+        cmd = agent.get_service_command("myservice", "start")
+        self.assertEqual(cmd, "systemctl start myservice")
+
+    @patch('agent.is_systemd', return_value=True)
+    def test_systemd_multiple_ops(self, _):
+        cmd = agent.get_service_command("myservice", "daemon-reload", "restart")
+        self.assertEqual(cmd, "systemctl daemon-reload myservice && systemctl restart myservice")
+
+    @patch('agent.is_systemd', return_value=False)
+    @patch('agent.hutil_log_info')
+    def test_initd_fallback(self, _, __):
+        cmd = agent.get_service_command("myservice", "start")
+        self.assertEqual(cmd, "/etc/init.d/myservice start")
+
+
+class TestConstants(unittest.TestCase):
+    """Tests for agent constants."""
+
+    def test_error_codes(self):
+        self.assertEqual(agent.GenericErrorCode, 1)
+        self.assertEqual(agent.UnsupportedOperatingSystem, 51)
+        self.assertEqual(agent.MissingorInvalidParameterErrorCode, 53)
+        self.assertEqual(agent.DPKGOrRPMLockedErrorCode, 56)
+        self.assertEqual(agent.MissingDependency, 52)
+
+    def test_supported_arch(self):
+        self.assertIn('x86_64', agent.SupportedArch)
+        self.assertIn('aarch64', agent.SupportedArch)
+
+    def test_config_keys(self):
+        self.assertEqual(agent.GenevaConfigKey, "genevaConfiguration")
+        self.assertEqual(agent.AzureMonitorConfigKey, "azureMonitorConfiguration")
+
+
+if __name__ == '__main__':
+    unittest.main()

AzureMonitorAgent/tests/test_agent.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python
+"""
+Unit tests for AzureMonitorAgent/ama_tst/modules/high_cpu_mem/check_logrot.py
+"""
+
+import sys
+import os
+import unittest
+from unittest.mock import patch
+
+# Add path for ama_tst modules
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'ama_tst', 'modules')))
+
+import error_codes
+from errors import error_info
+from high_cpu_mem.check_logrot import hr2bytes, check_size_config
+
+
+class TestHr2Bytes(unittest.TestCase):
+    """Tests for hr2bytes function."""
+
+    def test_plain_digits(self):
+        self.assertEqual(hr2bytes("1024"), 1024)
+
+    def test_kilobytes(self):
+        self.assertEqual(hr2bytes("5k"), 5000)
+
+    def test_megabytes(self):
+        self.assertEqual(hr2bytes("10M"), 10000000)
+
+    def test_gigabytes(self):
+        self.assertEqual(hr2bytes("2G"), 2000000000)
+
+    def test_invalid_unit(self):
+        self.assertIsNone(hr2bytes("5X"))
+
+    def test_invalid_format(self):
+        self.assertIsNone(hr2bytes("abc"))
+
+    def test_zero_kilobytes(self):
+        self.assertEqual(hr2bytes("0k"), 0)
+
+    def test_one_megabyte(self):
+        self.assertEqual(hr2bytes("1M"), 1000000)
+
+
+class TestCheckSizeConfig(unittest.TestCase):
+    """Tests for check_size_config function."""
+
+    def setUp(self):
+        error_info.clear()
+
+    def test_no_size_config(self):
+        configs = {"/var/log/test.log": {"rotate 5", "daily"}}
+        result = check_size_config(configs)
+        self.assertEqual(result, error_codes.NO_ERROR)
+
+    def test_invalid_size_format(self):
+        configs = {"/var/log/test.log": {"size XYZ", "rotate 5"}}
+        result = check_size_config(configs)
+        self.assertEqual(result, error_codes.ERR_LOGROTATE_SIZE)
+
+    @patch('os.path.getsize', return_value=500)
+    def test_file_under_limit(self, mock_size):
+        configs = {"/var/log/test.log": {"size 1M", "rotate 5"}}
+        result = check_size_config(configs)
+        self.assertEqual(result, error_codes.NO_ERROR)
+
+    @patch('os.path.getsize', return_value=2000000)
+    def test_file_over_limit(self, mock_size):
+        configs = {"/var/log/test.log": {"size 1M", "rotate 5"}}
+        result = check_size_config(configs)
+        self.assertEqual(result, error_codes.WARN_LOGROTATE)
+
+    @patch('os.path.getsize', side_effect=OSError(13, "Permission denied"))
+    def test_permission_denied(self, mock_size):
+        import errno
+        err = OSError(errno.EACCES, "Permission denied")
+        with patch('os.path.getsize', side_effect=err):
+            configs = {"/var/log/test.log": {"size 1M", "rotate 5"}}
+            result = check_size_config(configs)
+            self.assertEqual(result, error_codes.ERR_SUDO_PERMS)
+
+    @patch('os.path.getsize')
+    def test_file_not_found_missingok(self, mock_size):
+        import errno
+        mock_size.side_effect = OSError(errno.ENOENT, "No such file")
+        configs = {"/var/log/test.log": {"size 1M", "missingok"}}
+        result = check_size_config(configs)
+        self.assertEqual(result, error_codes.NO_ERROR)
+
+    @patch('os.path.getsize')
+    def test_file_not_found_no_missingok(self, mock_size):
+        import errno
+        mock_size.side_effect = OSError(errno.ENOENT, "No such file")
+        configs = {"/var/log/test.log": {"size 1M", "rotate 5"}}
+        result = check_size_config(configs)
+        self.assertEqual(result, error_codes.ERR_FILE_MISSING)
+
+
+if __name__ == '__main__':
+    unittest.main()