Run unit tests with pytest on Python >= 3.10

.github/workflows/ci_pr.yml

@@ -156,11 +156,14 @@ jobs:
 
         pylint $PYLINT_OPTIONS $PYLINT_FILES
 
-    - name: Test with nosetests
-      if: contains(fromJSON('["3.10", "3.11"]'), matrix.python-version) == false && (success() || (failure() && steps.install-dependencies.outcome == 'success'))
+    - name: Execute Unit Tests
+      if: success() || (failure() && steps.install-dependencies.outcome == 'success')
       run: |
-        ./ci/nosetests.sh
-        exit $?
+        if [[ "${{ matrix.python-version }}" =~ ^3\.[1-9][0-9]+$ ]]; then
+          ./ci/pytest.sh
+        else
+          ./ci/nosetests.sh
+        fi
 
     - name: Compile Coverage
       if: matrix.python-version == '3.9'

ci/nosetests.sh

@@ -5,18 +5,18 @@ set -u
 EXIT_CODE=0
 
 echo "========================================="
-echo "nosetests -a '!requires_sudo' output"
+echo "****     nosetests non-sudo tests    ****"
 echo "========================================="
-nosetests -a '!requires_sudo' tests $NOSEOPTS || EXIT_CODE=$(($EXIT_CODE || $?))
+nosetests --ignore-files test_cgroupconfigurator_sudo.py tests $NOSEOPTS || EXIT_CODE=$(($EXIT_CODE || $?))
 echo EXIT_CODE no_sudo nosetests = $EXIT_CODE
 
 [[ -f .coverage ]] && \
     sudo mv .coverage coverage.$(uuidgen).no_sudo.data
 
 echo "========================================="
-echo "nosetests -a 'requires_sudo' output"
+echo "****      nosetests sudo tests       ****"
 echo "========================================="
-sudo env "PATH=$PATH" nosetests -a 'requires_sudo' tests $NOSEOPTS || EXIT_CODE=$(($EXIT_CODE || $?))
+sudo env "PATH=$PATH" nosetests tests/ga/test_cgroupconfigurator_sudo.py $NOSEOPTS || EXIT_CODE=$(($EXIT_CODE || $?))
 echo EXIT_CODE with_sudo nosetests = $EXIT_CODE
 
 [[ -f .coverage ]] && \

ci/pytest.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+
+set -u
+
+EXIT_CODE=0
+
+echo "========================================="
+echo "****   pytest *** non-sudo tests     ****"
+echo "========================================="
+pytest --ignore-glob '*/test_cgroupconfigurator_sudo.py' --verbose tests || EXIT_CODE=$(($EXIT_CODE || $?))
+echo EXIT_CODE pytests non-sudo = $EXIT_CODE
+
+echo "========================================="
+echo "****     pytest *** sudo tests       ****"
+echo "========================================="
+sudo env "PATH=$PATH" pytest --verbose tests/ga/test_cgroupconfigurator_sudo.py || EXIT_CODE=$(($EXIT_CODE || $?))
+echo EXIT_CODE pytests sudo = $EXIT_CODE
+
+exit "$EXIT_CODE"

test-requirements.txt

@@ -3,8 +3,9 @@ mock==2.0.0; python_version == '2.6'
 mock==3.0.5; python_version >= '2.7' and python_version <= '3.5'
 mock==4.0.2; python_version >= '3.6'
 distro; python_version >= '3.8'
-nose
-nose-timer; python_version >= '2.7'
+nose; python_version <= '3.9'
+nose-timer; python_version >= '2.7' and python_version <= '3.9'
+pytest; python_version >= '3.10'
 
 # Pinning the wrapt requirement to 1.12.0 due to the bug - https://github.com/GrahamDumpleton/wrapt/issues/188
 wrapt==1.12.0; python_version > '2.6' and python_version < '3.6'

tests/common/test_logger.py

@@ -15,7 +15,6 @@
 # Requires Python 2.6+ and Openssl 1.0+
 #
 
-import json  # pylint: disable=unused-import
 import os
 import tempfile
 from datetime import datetime, timedelta
@@ -49,6 +48,7 @@ def tearDown(self):
         AgentTestCase.tearDown(self)
         logger.reset_periodic()
         logger.DEFAULT_LOGGER.appenders *= 0
+        logger.set_prefix(None)
         fileutil.rm_dirs(self.event_dir)
 
     @patch('azurelinuxagent.common.logger.Logger.verbose')

tests/common/test_telemetryevent.py

@@ -19,23 +19,23 @@
 from tests.lib.tools import AgentTestCase
 
 
-def get_test_event(name="DummyExtension", op="Unknown", is_success=True, duration=0, version="foo", evt_type="", is_internal=False,
-                      message="DummyMessage", eventId=1):
-    event = TelemetryEvent(eventId, "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX")
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Name, name))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Version, str(version)))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.IsInternal, is_internal))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Operation, op))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.OperationSuccess, is_success))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Message, message))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Duration, duration))
-    event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.ExtensionType, evt_type))
-    return event
-
-
 class TestTelemetryEvent(AgentTestCase):
+    @staticmethod
+    def _get_test_event(name="DummyExtension", op="Unknown", is_success=True, duration=0, version="foo", evt_type="", is_internal=False,
+                       message="DummyMessage", eventId=1):
+        event = TelemetryEvent(eventId, "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX")
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Name, name))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Version, str(version)))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.IsInternal, is_internal))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Operation, op))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.OperationSuccess, is_success))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Message, message))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.Duration, duration))
+        event.parameters.append(TelemetryEventParam(GuestAgentExtensionEventsSchema.ExtensionType, evt_type))
+        return event
+
     def test_contains_works_for_TelemetryEvent(self):
-        test_event = get_test_event(message="Dummy Event")
+        test_event = TestTelemetryEvent._get_test_event(message="Dummy Event")
 
         self.assertTrue(GuestAgentExtensionEventsSchema.Name in test_event)
         self.assertTrue(GuestAgentExtensionEventsSchema.Version in test_event)

tests/ga/test_cgroupconfigurator.py

@@ -26,20 +26,17 @@
 import time
 import threading
 
-from nose.plugins.attrib import attr
-
 from azurelinuxagent.common import conf
 from azurelinuxagent.ga.cgroup import AGENT_NAME_TELEMETRY, MetricsCounter, MetricValue, MetricsCategory, CpuCgroup
 from azurelinuxagent.ga.cgroupconfigurator import CGroupConfigurator, DisableCgroups
 from azurelinuxagent.ga.cgroupstelemetry import CGroupsTelemetry
 from azurelinuxagent.common.event import WALAEventOperation
-from azurelinuxagent.common.exception import CGroupsException, ExtensionError, ExtensionErrorCodes, \
-    AgentMemoryExceededException
+from azurelinuxagent.common.exception import CGroupsException, AgentMemoryExceededException
 from azurelinuxagent.common.future import ustr
 from azurelinuxagent.common.utils import shellutil, fileutil
 from tests.lib.mock_environment import MockCommand
 from tests.lib.mock_cgroup_environment import mock_cgroup_environment, UnitFilePaths
-from tests.lib.tools import AgentTestCase, patch, mock_sleep, i_am_root, data_dir, is_python_version_26_or_34, skip_if_predicate_true
+from tests.lib.tools import AgentTestCase, patch, mock_sleep, data_dir, is_python_version_26_or_34, skip_if_predicate_true
 from tests.lib.miscellaneous_tools import format_processes, wait_for
 
 
@@ -526,112 +523,6 @@ def test_start_extension_command_should_disable_cgroups_and_invoke_the_command_d
 
                         self.assertEqual(len(CGroupsTelemetry._tracked), 0, "No cgroups should have been created")
 
-    @skip_if_predicate_true(is_python_version_26_or_34, "Disabled on Python 2.6 and 3.4, they run on containers where the OS commands needed by the test are not present.")
-    @attr('requires_sudo')
-    @patch('time.sleep', side_effect=lambda _: mock_sleep())
-    def test_start_extension_command_should_not_use_fallback_option_if_extension_fails(self, *args):
-        self.assertTrue(i_am_root(), "Test does not run when non-root")
-
-        with self._get_cgroup_configurator() as configurator:
-            pass  # release the mocks used to create the test CGroupConfigurator so that they do not conflict the mock Popen below
-
-        command = "ls folder_does_not_exist"
-
-        with tempfile.TemporaryFile(dir=self.tmp_dir, mode="w+b") as stdout:
-            with tempfile.TemporaryFile(dir=self.tmp_dir, mode="w+b") as stderr:
-                with patch("azurelinuxagent.ga.cgroupapi.subprocess.Popen", wraps=subprocess.Popen) as popen_patch:
-                    with self.assertRaises(ExtensionError) as context_manager:
-                        configurator.start_extension_command(
-                            extension_name="Microsoft.Compute.TestExtension-1.2.3",
-                            command=command,
-                            cmd_name="test",
-                            timeout=300,
-                            shell=True,
-                            cwd=self.tmp_dir,
-                            env={},
-                            stdout=stdout,
-                            stderr=stderr)
-
-                    extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if command in args[0]]
-
-                    self.assertEqual(1, len(extension_calls), "The extension should have been invoked exactly once")
-                    self.assertIn("systemd-run", extension_calls[0],
-                                  "The first call to the extension should have used systemd")
-
-                    self.assertEqual(context_manager.exception.code, ExtensionErrorCodes.PluginUnknownFailure)
-                    self.assertIn("Non-zero exit code", ustr(context_manager.exception))
-                    # The scope name should appear in the process output since systemd-run was invoked and stderr
-                    # wasn't truncated.
-                    self.assertIn("Running scope as unit", ustr(context_manager.exception))
-
-    @skip_if_predicate_true(is_python_version_26_or_34, "Disabled on Python 2.6 and 3.4, they run on containers where the OS commands needed by the test are not present.")
-    @attr('requires_sudo')
-    @patch('time.sleep', side_effect=lambda _: mock_sleep())
-    @patch("azurelinuxagent.ga.extensionprocessutil.TELEMETRY_MESSAGE_MAX_LEN", 5)
-    def test_start_extension_command_should_not_use_fallback_option_if_extension_fails_with_long_output(self, *args):
-        self.assertTrue(i_am_root(), "Test does not run when non-root")
-
-        with self._get_cgroup_configurator() as configurator:
-            pass  # release the mocks used to create the test CGroupConfigurator so that they do not conflict the mock Popen below
-
-        long_output = "a"*20  # large enough to ensure both stdout and stderr are truncated
-        long_stdout_stderr_command = "echo {0} && echo {0} >&2 && ls folder_does_not_exist".format(long_output)
-
-        with tempfile.TemporaryFile(dir=self.tmp_dir, mode="w+b") as stdout:
-            with tempfile.TemporaryFile(dir=self.tmp_dir, mode="w+b") as stderr:
-                with patch("azurelinuxagent.ga.cgroupapi.subprocess.Popen", wraps=subprocess.Popen) as popen_patch:
-                    with self.assertRaises(ExtensionError) as context_manager:
-                        configurator.start_extension_command(
-                            extension_name="Microsoft.Compute.TestExtension-1.2.3",
-                            command=long_stdout_stderr_command,
-                            cmd_name="test",
-                            timeout=300,
-                            shell=True,
-                            cwd=self.tmp_dir,
-                            env={},
-                            stdout=stdout,
-                            stderr=stderr)
-
-                    extension_calls = [args[0] for (args, _) in popen_patch.call_args_list if long_stdout_stderr_command in args[0]]
-
-                    self.assertEqual(1, len(extension_calls), "The extension should have been invoked exactly once")
-                    self.assertIn("systemd-run", extension_calls[0],
-                                  "The first call to the extension should have used systemd")
-
-                    self.assertEqual(context_manager.exception.code, ExtensionErrorCodes.PluginUnknownFailure)
-                    self.assertIn("Non-zero exit code", ustr(context_manager.exception))
-                    # stdout and stderr should have been truncated, so the scope name doesn't appear in stderr
-                    # even though systemd-run ran
-                    self.assertNotIn("Running scope as unit", ustr(context_manager.exception))
-
-    @attr('requires_sudo')
-    def test_start_extension_command_should_not_use_fallback_option_if_extension_times_out(self, *args):  # pylint: disable=unused-argument
-        self.assertTrue(i_am_root(), "Test does not run when non-root")
-
-        with self._get_cgroup_configurator() as configurator:
-            pass  # release the mocks used to create the test CGroupConfigurator so that they do not conflict the mock Popen below
-
-        with tempfile.TemporaryFile(dir=self.tmp_dir, mode="w+b") as stdout:
-            with tempfile.TemporaryFile(dir=self.tmp_dir, mode="w+b") as stderr:
-                with patch("azurelinuxagent.ga.extensionprocessutil.wait_for_process_completion_or_timeout",
-                           return_value=[True, None, 0]):
-                    with patch("azurelinuxagent.ga.cgroupapi.SystemdCgroupsApi._is_systemd_failure",
-                               return_value=False):
-                        with self.assertRaises(ExtensionError) as context_manager:
-                            configurator.start_extension_command(
-                                extension_name="Microsoft.Compute.TestExtension-1.2.3",
-                                command="date",
-                                cmd_name="test",
-                                timeout=300,
-                                shell=True,
-                                cwd=self.tmp_dir,
-                                env={},
-                                stdout=stdout,
-                                stderr=stderr)
-
-                        self.assertEqual(context_manager.exception.code, ExtensionErrorCodes.PluginHandlerScriptTimedout)
-                        self.assertIn("Timeout", ustr(context_manager.exception))
-
     @skip_if_predicate_true(is_python_version_26_or_34, "Disabled on Python 2.6 and 3.4, they run on containers where the OS commands needed by the test are not present.")
     @patch('time.sleep', side_effect=lambda _: mock_sleep())
     def test_start_extension_command_should_capture_only_the_last_subprocess_output(self, _):
@@ -1024,4 +915,4 @@ def test_check_agent_memory_usage_should_raise_a_cgroups_exception_when_the_limi
                     tracked_metrics.return_value = metrics
                     configurator.check_agent_memory_usage()
 
-        self.assertIn("The agent memory limit {0} bytes exceeded".format(conf.get_agent_memory_quota()), ustr(context_manager.exception), "An incorrect exception was raised")
+        self.assertIn("The agent memory limit {0} bytes exceeded".format(conf.get_agent_memory_quota()), ustr(context_manager.exception), "An incorrect exception was raised")