Skip to content

Commit

Permalink
add unit testing to increase code coverage (#265)
Browse files Browse the repository at this point in the history
* add raise exception utest in maintenance windows file

* add ut for log_to_stdout to true

* add test code to assert log output

* add raise exception test in Test_ConfigurePatchingProcessor.py

* add unit test for raise exception on auto assessment systemd

* refactor python file
  • Loading branch information
feng-j678 authored Sep 19, 2024
1 parent c01c7bc commit 0824ba8
Show file tree
Hide file tree
Showing 2 changed files with 101 additions and 0 deletions.
62 changes: 62 additions & 0 deletions src/core/tests/Test_ConfigurePatchingProcessor.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@
import os
import re
import unittest
import sys
from io import StringIO

from core.src.CoreMain import CoreMain
from core.src.bootstrap.Constants import Constants
from core.tests.library.ArgumentComposer import ArgumentComposer
Expand All @@ -42,6 +45,8 @@ def mock_package_manager_get_current_auto_os_patch_state_returns_unknown(self):
return Constants.AutomaticOSPatchStates.DISABLED
else:
return Constants.AutomaticOSPatchStates.UNKNOWN
def mock_get_current_auto_os_patch_state(self):
raise Exception("Mocked Exception")
#endregion Mocks

def test_operation_success_for_configure_patching_request_for_apt_with_default_updates_config(self):
Expand Down Expand Up @@ -309,6 +314,63 @@ def test_configure_patching_with_patch_mode_and_assessment_mode_by_platform(self
# stop test runtime
runtime.stop()

def test_configure_patching_raise_exception_auto_os_patch_state(self):
# arrange capture std IO
captured_output = StringIO()
sys.stdout = captured_output

argument_composer = ArgumentComposer()
argument_composer.operation = Constants.CONFIGURE_PATCHING
argument_composer.patch_mode = Constants.PatchModes.AUTOMATIC_BY_PLATFORM
argument_composer.assessment_mode = Constants.AssessmentModes.AUTOMATIC_BY_PLATFORM
runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
runtime.package_manager.get_current_auto_os_patch_state = runtime.backup_get_current_auto_os_patch_state
runtime.set_legacy_test_type('HappyPath')

# mock swap
backup_package_manager_get_current_auto_os_patch_state = runtime.package_manager.get_current_auto_os_patch_state
runtime.package_manager.get_current_auto_os_patch_state = self.mock_get_current_auto_os_patch_state

runtime.configure_patching_processor.start_configure_patching()

# restore sdt.out ouptput
sys.stdout = sys.__stdout__

# assert
output = captured_output.getvalue()
self.assertIn("Error while processing patch mode configuration", output)

# check status file
with runtime.env_layer.file_system.open(runtime.execution_config.status_file_path, 'r') as file_handle:
substatus_file_data = json.load(file_handle)[0]["status"]["substatus"]
self.assertEqual(len(substatus_file_data), 1)
self.assertTrue(substatus_file_data[0]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY)
self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_TRANSITIONING.lower())

# restore
runtime.package_manager.get_current_auto_os_patch_state = backup_package_manager_get_current_auto_os_patch_state

runtime.stop()

def test_configure_patching_raise_exception_auto_assessment_systemd(self):
argument_composer = ArgumentComposer()
argument_composer.operation = Constants.CONFIGURE_PATCHING
argument_composer.patch_mode = Constants.PatchModes.AUTOMATIC_BY_PLATFORM
argument_composer.assessment_mode = Constants.AssessmentModes.AUTOMATIC_BY_PLATFORM
runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
runtime.set_legacy_test_type('HappyPath')

# mock swap
back_up_auto_assess_service_manager = runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists
runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists = lambda: False

self.assertRaises(Exception, runtime.configure_patching_processor.start_configure_patching())

# restore
runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists = back_up_auto_assess_service_manager

runtime.stop()

def __check_telemetry_events(self, runtime):
all_events = os.listdir(runtime.telemetry_writer.events_folder_path)
self.assertTrue(len(all_events) > 0)
Expand Down
39 changes: 39 additions & 0 deletions src/core/tests/Test_MaintenanceWindow.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
# Requires Python 2.7+

import datetime
from io import StringIO
import sys
import unittest
from core.tests.library.ArgumentComposer import ArgumentComposer
from core.tests.library.RuntimeCompositor import RuntimeCompositor
Expand Down Expand Up @@ -62,6 +64,43 @@ def test_RemainingTime_after_duration_complete(self):
self.assertEqual(int(remaining_time), 0)
runtime.stop()

def test_RemainingTime_log_to_stdout_true(self):
# Arrange, Capture stdout
captured_output = StringIO()
sys.stdout = captured_output # Redirect stdout to the StringIO object

argument_composer = ArgumentComposer()
argument_composer.start_time = "2017-02-15T18:15:12.9828835Z"
argument_composer.maximum_duration = "PT1H"
runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True)

current_time = datetime.datetime.strptime('2017-02-15 18:30:20', "%Y-%m-%d %H:%M:%S")
remaining_time = runtime.maintenance_window.get_remaining_time_in_minutes(current_time, log_to_stdout=True)

# Restore stdout
sys.stdout = sys.__stdout__

# Assert
output = captured_output.getvalue()
self.assertEqual(int(remaining_time), 44)
self.assertIn("Maintenance Window Utilization:", output) # Verify the log output contains the expected text

runtime.stop()

def test_RemainingTime_raise_exception(self):
# Arrange
argument_composer = ArgumentComposer()
argument_composer.start_time = "Invalid datetime format"
argument_composer.maximum_duration = "PT1H"
runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True)

# Assert
with self.assertRaises(ValueError) as context:
runtime.maintenance_window.get_remaining_time_in_minutes()
self.assertIn("Invalid datetime format", str(context.exception))

runtime.stop()

def test_check_available_time(self):
argument_composer = ArgumentComposer()
argument_composer.start_time = (datetime.datetime.utcnow() - datetime.timedelta(minutes=39)).strftime("%Y-%m-%dT%H:%M:%S.9999Z")
Expand Down

0 comments on commit 0824ba8

Please sign in to comment.